对于PostgreSQL的pgmq扩展，这里有简介而又完整的介绍：
1，pgmq是一个基于 PostgreSQL 的轻量级消息队列扩展程序。其 API 设计与亚马逊 SQS 和 RSMQ 类似，为用户提供了简单而强大的消息处理能力。
2，它完全通过 SQL/PLpgSQL 实现，无需任何后端工作进程或外部依赖，极大地简化了部署和维护工作。
3，pgmq采用“队列即表”的架构，将消息存储在常规的 JSONB 格式的表中。
4，它支持可见性超时、延迟交付、长轮询、批量操作（发送/归档/删除）以及丰富的监控指标，使其成为构建事件驱动架构、处理后台任务和解耦微服务的理想选择。
pgmq提供了高性能，高可靠，轻量级，功能全面的消息处理功能，本文基于github上的介绍以及文档，粗略测试与体验pgmq的功能。
每个队列在 pgmq 中对应一个独立的表。表名是队列名加前缀 q_，例如：队列 my_queue 对应的表为 pgmq.q_my_queue，同时对应一个归档的表，名为pgmq.a_my_queue，由于create是一个function，其代码很非常清晰。
-- 创建队列
SELECT pgmq.create('my_queue');
create
-------------
(1 row)消除的生成通过pgmq.send来实现，实际上是将消息写入创建消息时候生成的表中
--消息以 JSON 格式发送，pgmq.send 会返回消息 ID
-- 发送第一条消息
SELECT * FROM pgmq.send(
queue_name => 'my_queue',
msg => '{"foo": "bar1"}'
);
send
-----------
1
(1 row)
可以选择指定延迟，这条消息会在队列中，但在 5 秒内不可被消费：
SELECT * FROM pgmq.send(
queue_name => 'my_queue',
msg => '{"foo": "bar2"}',
delay => 5
);
send
-----------
2
(1 row)这里是所以用“读取”而不是消费，是因为这个单纯的就是一个查询消息的行为，与消息的弹出（pop）相比，不涉及“消费”，读取完消息还在队列之中，消息本身不会发生变化。
正常读取（read）
--读取队列中的消息并设置可见性超时 (vt)
--如果消息在 30 秒内未被删除或归档，会重新可见给其他消费者
SELECT * FROM pgmq.read(
queue_name => 'my_queue',
vt => 30,
qty => 2
);
msg_id | read_ct | enqueued_at | last_read_at | vt | message | headers
--------+---------+-------------------------------+-------------------------------+-------------------------------+-----------------+---------
1 | 1 | 2026-01-23 20:27:21.7741-06 | 2026-01-23 20:27:31.605236-06 | 2026-01-23 20:28:01.605236-06 | {"foo": "bar1"} |
2 | 1 | 2026-01-23 20:27:26.505063-06 | 2026-01-23 20:27:31.605252-06 | 2026-01-23 20:28:01.605252-06 | {"foo": "bar2"} |
--如果队列为空，或者所有消息当前不可见，则返回空行：
SELECT * FROM pgmq.read(
queue_name => 'my_queue',
vt => 30,
qty => 1
);
msg_id | read_ct | enqueued_at | last_read_at | vt | message | headers
--------+---------+-------------+--------------+----+---------+---------阻塞的方式去读取消息，也即通过等待来获取最新的消息，需要用到read_with_poll函数，该函数的详细参数以及使用方式如下，值得注意的是poll_interval_ms参数，如果轮询间隔（毫秒）太短，会导致服务器CPU暴涨的情况。
read_with_poll函数的参数：
/*
Parameter Type Description
queue_name text The name of the queue
vt integer Time in seconds that the message become invisible after reading.
qty integer The number of messages to read from the queue
max_poll_seconds integer Time in seconds to wait for new messages to reach the queue. Defaults to 5.
poll_interval_ms integer Milliseconds between the internal poll operations. Defaults to 100.
conditional jsonb Filters the messages by their json content. Defaults to '{}' - no filtering. This feature is experimental, and the API is subject to change in future releases
*/
-- 从my_queue读取1条消息，10秒钟之内消息不可见，等待5秒钟，内部循环时间为100毫秒
SELECT * from pgmq.read_with_poll(queue_name =>'my_queue',
vt =>10,
qty =>1,
max_poll_seconds=>5,
poll_interval_ms=>100)
该方式有多个类似的方法：read_grouped_rr_with_poll，read_grouped_with_poll
消息的弹出（pop）与消息的读取有着本质的不同，消息被弹出之后，相关的消息立即从队列中删除。
SELECT * FROM pgmq.pop('my_queue');
msg_id | read_ct | enqueued_at | last_read_at | vt | message | headers
--------+---------+-----------------------------+-------------------------------+-------------------------------+-----------------+---------
1 | 1 | 2026-01-23 20:27:21.7741-06 | 2026-01-23 20:27:31.605236-06 | 2026-01-23 20:28:01.605236-06 | {"foo": "bar1"} |在队列的创建中提到了，创建队列的时候会自动创建与队列名字相关的两张表，归档消息，就是将消息从队列中删除，并插入到归档表，可以单条归档或者批量归档。
-- 归档消息 msg_id=2
SELECT pgmq.archive(
queue_name => 'my_queue',
msg_id => 2
);
archive
--------------
t
(1 row)
-- 批量归档
SELECT pgmq.archive(
queue_name => 'my_queue',
msg_ids => ARRAY[3, 4, 5]
);
archive
---------
3
4
5
查看归档表：
SELECT * FROM pgmq.a_my_queue;
msg_id | read_ct | enqueued_at | last_read_at | archived_at | vt | message | headers
--------+---------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+---------
2 | 1 | 2026-01-23 20:32:35.291971-06 | 2026-01-23 20:32:42.938473-06 | 2026-01-23 20:33:20.297454-06 | 2026-01-23 20:33:12.938473-06 | {"foo": "bar2"} |
3 | 0 | 2026-01-23 20:33:25.414914-06 | | 2026-01-23 20:33:30.318465-06 | 2026-01-23 20:33:25.415035-06 | {"foo": "bar3"} |
4 | 0 | 2026-01-23 20:33:25.414914-06 | | 2026-01-23 20:33:30.318465-06 | 2026-01-23 20:33:25.415035-06 | {"foo": "bar4"} |
5 | 0 | 2026-01-23 20:33:25.414914-06 | | 2026-01-23 20:33:30.318465-06 | 2026-01-23 20:33:25.415035-06 | {"foo": "bar5"} |与消息归档不同的是，消息的删除是一个纯粹的删除动作，直接从队列（或者叫消息表中）删除对应Id的数据
SELECT pgmq.delete('my_queue', 6);
delete
-------------
t
(1 row)SELECT pgmq.drop_queue('my_queue');
drop_queue
-----------------
t
(1 row)
PGMQ FIFO 队列的特点是返回每个组的最早的一条消息，其特点是公平消费每个组的消息，如下，队列的创建一些消息的生成与上述一样，只是在消费端用到了新的方式
SELECT pgmq.create('order_processing');
-- read_grouped_rr的逻辑是分组后读取每个组的最早的消息，因此创建一个索引来增加查询效率
SELECT pgmq.create_fifo_index('order_processing');
-- Example 1: Basic FIFO ordering within a customer group
-- Send multiple orders for the same customer - these must be processed in order
SELECT pgmq.send('order_processing',
'{"customer_id": "cust_123", "order_id": "ord_001", "action": "create", "amount": 100}'::jsonb,
'{"x-pgmq-group": "cust_123"}'::jsonb
);
SELECT pgmq.send('order_processing',
'{"customer_id": "cust_123", "order_id": "ord_001", "action": "update", "amount": 150}'::jsonb,
'{"x-pgmq-group": "cust_123"}'::jsonb
);
SELECT pgmq.send('order_processing',
'{"customer_id": "cust_123", "order_id": "ord_001", "action": "complete"}'::jsonb,
'{"x-pgmq-group": "cust_123"}'::jsonb
);
-- Example 2: Parallel processing for different customers
-- Send orders for different customers - these can be processed in parallel
SELECT pgmq.send('order_processing',
'{"customer_id": "cust_456", "order_id": "ord_002", "action": "create", "amount": 200}'::jsonb,
'{"x-pgmq-group": "cust_456"}'::jsonb
);
SELECT pgmq.send('order_processing',
'{"customer_id": "cust_789", "order_id": "ord_003", "action": "create", "amount": 300}'::jsonb,
'{"x-pgmq-group": "cust_789"}'::jsonb
);
select pgmq.read(queue_name => 'order_processing',
vt => 1,
qty => 100)
read |
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
(2,2,'2026-05-08 19:44:59.868534+08','2026-05-08 19:46:09.860615+08','2026-05-08 19:46:10.860615+08',{"action": "update", "amount": 150, "order_id": "ord_001", "customer_id": "cust_123"},{"x-pgmq-group": "cust_123"})|
(3,2,'2026-05-08 19:45:03.52429+08','2026-05-08 19:46:09.860678+08','2026-05-08 19:46:10.860678+08',{"action": "complete", "order_id": "ord_001", "customer_id": "cust_123"},{"x-pgmq-group": "cust_123"}) |
(5,2,'2026-05-08 19:45:06.166905+08','2026-05-08 19:46:09.860686+08','2026-05-08 19:46:10.860686+08',{"action": "create", "amount": 300, "order_id": "ord_003", "customer_id": "cust_789"},{"x-pgmq-group": "cust_789"})|
(4,3,'2026-05-08 19:45:04.925428+08','2026-05-08 19:46:09.860692+08','2026-05-08 19:46:10.860692+08',{"action": "create", "amount": 200, "order_id": "ord_002", "customer_id": "cust_456"},{"x-pgmq-group": "cust_456"})|
(1,4,'2026-05-08 19:44:58.268758+08','2026-05-08 19:46:09.860698+08','2026-05-08 19:46:10.860698+08',{"action": "create", "amount": 100, "order_id": "ord_001", "customer_id": "cust_123"},{"x-pgmq-group": "cust_123"})|
这里返回每个组的第一条消息
SELECT
msg_id,
message->>'customer_id' as customer_id,
message->>'action' as action,
headers->>'x-pgmq-group' as fifo_group
FROM pgmq.read_grouped_rr(
queue_name => 'order_processing',
vt => 1,
qty => 2);
msg_id|customer_id|action|fifo_group|
------+-----------+------+----------+
1|cust_123 |create|cust_123 |
4|cust_456 |create|cust_456 |根据FIFO的原则返回满足请求的消息的条数，Should attempt to return as many messages as possible from the same group，直白地说就是没有了分组（group）的概念，消息不看归属的组（group），按照进入队列的顺序来读取
SELECT
'Round-Robin' as test_type,
msg_id,
message->>'group' as group_id,
message->>'message_num' as message_num,
headers->>'x-pgmq-group' as fifo_group
FROM pgmq.read_grouped(queue_name => 'order_processing',
vt => 1,
qty => 3)
ORDER BY msg_id;
由于cust_123这个组的消息是最早进入队列的，且已经满足qty=3，因此这里返回的是cust_123这个组的全部数据
test_type |msg_id|group_id|message_num|fifo_group|
-----------+------+--------+-----------+----------+
Round-Robin| 1| | |cust_123 |
Round-Robin| 2| | |cust_123 |
Round-Robin| 3| | |cust_123 |类似于消息的多播，队列与主题关联， 让多个队列“订阅”或“绑定”到一个 topic，之后向这个 topic 发送的消息会进入绑定的队列，每个队列可以独立消费，互不干扰，从而实现多队列消息分发
队列与主题关联， 让队列“订阅”或“绑定”到一个 topic,
SELECT pgmq.create('my_queue1');
SELECT pgmq.create('my_queue2');
SELECT pgmq.bind_topic('my_topic', 'my_queue1');
SELECT pgmq.bind_topic('my_topic', 'my_queue2');
之后向这个 topic 发送的消息会进入绑定的队列
SELECT pgmq.send_topic('my_topic', '{"msg": "message info1"}'::jsonb, NULL, 0);
SELECT pgmq.send_topic('my_topic', '{"msg": "message info2"}'::jsonb, NULL, 0);
可以有多个队列绑定同一个 topic，每个队列可以独立消费，互不干扰，从而实现多队列消息分发
这里可以看到，由于my_queue1和my_queue2都绑定到了my_topic上，然后send_topic的时候，会同时将消息发送给两个队列。
SELECT * FROM pgmq.read(
queue_name => 'my_queue1',
vt => 1,
qty => 2
) order by msg_id;
msg_id|read_ct|enqueued_at |last_read_at |vt |message |headers|
------+-------+--------------------------------+--------------------------------+--------------------------------+------------------------+-------+
1| 4|2026-05-08 19:35:04.898635 +0800|2026-05-08 19:38:55.932311 +0800|2026-05-08 19:38:56.932311 +0800|{"msg": "message info"} | |
2| 4|2026-05-08 19:35:51.307749 +0800|2026-05-08 19:38:55.932338 +0800|2026-05-08 19:38:56.932338 +0800|{"msg": "message info1"}| |
SELECT * FROM pgmq.read(
queue_name => 'my_queue2',
vt => 0,
qty => 2
) order by msg_id;
msg_id|read_ct|enqueued_at |last_read_at |vt |message |headers|
------+-------+--------------------------------+--------------------------------+--------------------------------+------------------------+-------+
1| 2|2026-05-08 19:35:04.898635 +0800|2026-05-08 19:39:07.948205 +0800|2026-05-08 19:39:07.948205 +0800|{"msg": "message info"} | |
2| 2|2026-05-08 19:35:51.307749 +0800|2026-05-08 19:39:07.948237 +0800|2026-05-08 19:39:07.948237 +0800|{"msg": "message info1"}| |github上有压测的python脚本，笔者没有做详细的压力测试，在相关的材料中找到了一些2023年的pgmq 1.0压测结果，这个是在16C+32G的环境中测试的。
如下只是一个最小化key的测试结果，读可以3W/S每秒，写可以达到15W/S，因为视频中有多个不同场景的测试结果，这里不一一截图， 具体在这里：https://www.youtube.com/watch?v=a-jHFyEwVrA&t=256s
可见pgmq在中小负载的队列场景下还是可以胜任的。
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。