pgmq

pig build pkg pgmq         # 构建 RPM / DEB 包

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

CREATE EXTENSION pgmq;

CREATE EXTENSION pgmq;

SELECT pgmq.create('my_queue');

-- 发送单条消息（返回 msg_id）
SELECT * FROM pgmq.send(
  queue_name => 'my_queue',
  msg        => '{"foo": "bar"}'
);

-- 延迟发送（5 秒内不可见）
SELECT * FROM pgmq.send(
  queue_name => 'my_queue',
  msg        => '{"foo": "bar"}',
  delay      => 5
);

-- 批量发送消息
SELECT pgmq.send_batch(
  queue_name => 'my_queue',
  msgs       => ARRAY['{"a":1}','{"b":2}','{"c":3}']::jsonb[]
);

SELECT * FROM pgmq.read(
  queue_name => 'my_queue',
  vt         => 30,    -- 可见性超时（秒）
  qty        => 2      -- 读取消息数量
);

SELECT * FROM pgmq.pop('my_queue');

SELECT pgmq.delete('my_queue', 6);

SELECT pgmq.archive(queue_name => 'my_queue', msg_id => 2);

-- 批量归档
SELECT pgmq.archive(queue_name => 'my_queue', msg_ids => ARRAY[3, 4, 5]);

SELECT * FROM pgmq.a_my_queue;

SELECT pgmq.drop_queue('my_queue');