pgq

pig repo add pgdg -u          # 添加 PGDG 仓库并更新缓存

CREATE EXTENSION pgq;

CREATE EXTENSION pgq;

-- 创建队列
SELECT pgq.create_queue('myqueue');

-- 删除队列
SELECT pgq.drop_queue('myqueue');

-- 获取队列信息
SELECT * FROM pgq.get_queue_info();
SELECT * FROM pgq.get_queue_info('myqueue');

-- 在队列上注册消费者
SELECT pgq.register_consumer('myqueue', 'myconsumer');

-- 注销消费者
SELECT pgq.unregister_consumer('myqueue', 'myconsumer');

-- 获取消费者信息
SELECT * FROM pgq.get_consumer_info('myqueue');

-- 向队列插入事件
SELECT pgq.insert_event('myqueue', 'event_type', 'event_data');

-- 插入带额外字段的事件
SELECT pgq.insert_event('myqueue', 'event_type', 'event_data',
                         'extra1', 'extra2', 'extra3', 'extra4');

-- 获取下一批事件（返回 batch_id，无新批次时返回 NULL）
SELECT pgq.next_batch('myqueue', 'myconsumer');

-- 从批次中获取事件
SELECT * FROM pgq.get_batch_events(:batch_id);

-- 重试失败的事件（在指定间隔后重新出现）
SELECT pgq.event_retry(:batch_id, :event_id, :retry_seconds);

-- 标记批次完成
SELECT pgq.finish_batch(:batch_id);

-- 1. 获取下一批次
SELECT pgq.next_batch('myqueue', 'myconsumer') AS batch_id;

-- 2. 如果 batch_id 不为 NULL，获取事件
SELECT * FROM pgq.get_batch_events(:batch_id);

-- 3. 处理事件，重试失败的
SELECT pgq.event_retry(:batch_id, :event_id, 60);

-- 4. 完成批次
SELECT pgq.finish_batch(:batch_id);