文档中心

云数据库 PostgreSQL 版

插件

原生消息队列（pgmq）

简介

pgmq 是一款基于 PostgreSQL 实现的轻量级消息队列扩展，其 API 设计与 Amazon SQS 和 RSMQ 类似，为用户提供了简单而强大的消息处理能力。它完全通过 SQL/PLpgSQL 实现，无需任何后台工作进程或外部依赖，极大地简化了部署和维护。

pgmq 采用“队列即表”的架构，将消息以 JSONB 格式存储在普通表中。它支持可见性超时、延迟投递、长轮询、批量操作（发送/归档/删除）以及丰富的监控度量，是构建事件驱动架构、处理后台任务和解耦微服务的理想选择。

适用场景

异步任务处理：将耗时操作（如发送邮件、生成报告、图像处理）放入队列，由后台消费者异步处理，提升系统响应速度。

服务解耦：在微服务架构中，通过消息队列传递事件，降低服务间的直接依赖。

数据管道：作为轻量级数据管道，用于在不同系统或处理阶段之间可靠地传递数据。

延迟任务：利用延迟投递功能，实现定时提醒、定时任务触发等功能。

创建与删除插件

创建插件

执行以下 SQL 命令以启用 pgmq 插件。

删除插件

如果不再需要，可以执行以下命令将其卸载。卸载插件会删除所有相关的队列、消息和函数。

基础使用

队列管理

创建队列（pgmq.create）

创建一个新的消息队列。每个队列在物理上对应一个数据表。

-- 创建一个名为 my_queue 的普通队列
SELECT pgmq.create('my_queue');

删除队列（pgmq.drop_queue）

删除一个消息队列及其所有关联的数据表（包括主队列和归档队列）。

-- 删除 my_queue 消息队列
SELECT pgmq.drop_queue('my_queue');

消息操作

发送消息（pgmq.send）

向指定队列发送一条消息。

queue_name (text): 必需，目标队列的名称。

msg (jsonb): 必需，要发送的消息体，必须是 JSONB 格式。

delay (integer): 可选，消息的延迟投递时间（秒）。消息将在指定延迟后才对消费者可见。

-- 发送一条立即投递的消息
SELECT * FROM pgmq.send(
queue_name => 'my_queue',
msg => '{"task": "process_order", "order_id": 101}'
);

-- 发送一条 10 秒后才可见的延迟消息
SELECT * FROM pgmq.send(
queue_name => 'my_queue',
msg => '{"task": "send_reminder", "user_id": 205}',
delay => 10
);

批量发送消息（pgmq.send_batch）

一次性向队列发送多条消息，以提升性能。

msgs (jsonb[]): 必需，一个 JSONB 数组，包含多条消息。

SELECT pgmq.send_batch(
queue_name => 'my_queue',
msgs => ARRAY[
'{"task": "batch_job", "job_id": 1}',
'{"task": "batch_job", "job_id": 2}'
]::jsonb[]
);

读取消息（pgmq.read）

从队列中读取一批消息，但并不立即删除它们。读取后，消息会进入“不可见”状态，直到可见性超时（vt）结束。

vt (integer): 必需，可见性超时时间（秒）。在此期间，其他消费者无法读取到这些消息。

qty (integer): 必需，要读取的最大消息数量。

-- 从队列中读取最多 5 条消息，并设置 60 秒的可见性超时
SELECT * FROM pgmq.read(
queue_name => 'my_queue',
vt => 60,
qty => 5
);

如果成功读取，函数将返回一个包含消息详情的集合，包括 msg_id、message 等。如果在指定时间内没有新消息，长轮询模式下会等待，否则返回空集。

弹出消息（pgmq.pop）

pop 操作将读取和删除消息合并为原子一步。它会从队列中读取一条消息并立即将其删除。

-- 读取并删除一条消息
SELECT * FROM pgmq.pop('my_queue');

这适用于不需要消息确认机制的简单场景。

归档消息（pgmq.archive）

处理完成后，将消息从主队列移动到归档队列中。这对于审计和问题追溯非常有用。归档表以 a_ 为前缀。

msg_id (bigint)/msg_ids (bigint[]): 必需，要归档的单条或多条消息 ID。

-- 归档单条消息
SELECT pgmq.archive('my_queue', 123);

-- 批量归档多条消息
SELECT pgmq.archive('my_queue', ARRAY[124, 125, 126]);

归档后，可以查询归档表 pgmq.a_my_queue 来查看历史消息。

删除消息（pgmq.delete）

永久删除一条或多条消息，这些消息将不会被归档。

-- 删除单条消息
SELECT pgmq.delete('my_queue', 127);

-- 批量删除多条消息
SELECT pgmq.delete('my_queue', ARRAY[128, 129]);

监控与度量

pgmq 提供了视图和函数来监控队列的状态。

pgmq.metrics_all() 视图：提供所有队列的实时度量，如队列名称、消息总数、可见消息数、不可见消息数等。

SELECT * FROM pgmq.metrics_all();

pgmq.metrics(queue_name) 函数：获取指定单个队列的详细度量。

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

最近更新时间：2026.03.31 10:39:12

这个页面对您有帮助吗？

有用

无用