PGMQ 扩展

pgmq 是一个基于 Postgres 构建的轻量级消息队列。

🌐 pgmq is a lightweight message queue built on Postgres.

功能 #

🌐 Features

轻量级 - 没有后台工作程序或外部依赖，只是打包在扩展中的 Postgres 函数
在可见性超时内向消费者“准确一次”传递消息
与 AWS SQS 和 RSMQ 的 API 一致性
消息会一直留在队列里，直到被明确删除
消息可以被归档，而不是删除，以便长期保存和再次播放

启用扩展 #

🌐 Enable the extension

1
create extension pgmq;

使用情况 #

🌐 Usage [#get-usage]

排队管理 #

🌐 Queue management

`create`#

创建一个新队列。

🌐 Create a new queue.

1
pgmq.create(queue_name text)
2
returns void

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

1
select from pgmq.create('my_queue');
2
 create
3
--------

`create_unlogged`#

创建一个不记录日志的表。当写入吞吐量比持久性更重要时，这很有用。有关更多信息，请参见 Postgres 文档中的不记录日志的表。

🌐 Creates an unlogged table. This is useful when write throughput is more important than durability. See Postgres documentation for unlogged tables for more information.

1
pgmq.create_unlogged(queue_name text)
2
returns void

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

1
select pgmq.create_unlogged('my_unlogged');
2
 create_unlogged
3
-----------------

`detach_archive`#

作为 PGMQ 扩展的一部分，删除队列的归档表。这对于防止在执行 drop extension pgmq 时队列的归档表被删除很有用。这并不能阻止后续的 archives() 继续追加到归档表中。

🌐 Drop the queue's archive table as a member of the PGMQ extension. Useful for preventing the queue's archive table from being dropped when drop extension pgmq is executed. This does not prevent the further archives() from appending to the archive table.

1
pgmq.detach_archive(queue_name text)

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

1
select * from pgmq.detach_archive('my_queue');
2
 detach_archive
3
----------------

`drop_queue`#

删除一个队列及其归档表。

🌐 Deletes a queue and its archive table.

1
pgmq.drop_queue(queue_name text)
2
returns boolean

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

1
select * from pgmq.drop_queue('my_unlogged');
2
 drop_queue
3
------------
4
 t

发送消息 #

🌐 Sending messages

`send`#

向队列发送一条消息。

🌐 Send a single message to a queue.

1
pgmq.send(
2
    queue_name text,
3
    msg jsonb,
4
    delay integer default 0
5
)
6
returns setof bigint

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg`	`jsonb`	要发送到队列的消息
`delay`	`integer`	消息变为可见前的时间（秒）。默认是 0。

示例：

🌐 Example:

1
select * from pgmq.send('my_queue', '{"hello": "world"}');
2
 send
3
------
4
    4

`send_batch`#

向队列发送一条或多条消息。

🌐 Send 1 or more messages to a queue.

1
pgmq.send_batch(
2
    queue_name text,
3
    msgs jsonb[],
4
    delay integer default 0
5
)
6
returns setof bigint

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msgs`	`jsonb[]`	要发送到队列的消息数组
`delay`	`integer`	消息变为可见前的时间（秒）。默认值为 0。

1
select * from pgmq.send_batch(
2
    'my_queue',
3
    array[
4
      '{"hello": "world_0"}'::jsonb,
5
      '{"hello": "world_1"}'::jsonb
6
    ]
7
);
8
 send_batch
9
------------
10
          1
11
          2

看消息 #

🌐 Reading messages

`read`#

从队列中读取一条或多条消息。VT 指定消息对其他消费者不可见的时间（以秒为单位）。该时间结束后，消息会再次可见，并可以被其他消费者读取。

🌐 Read 1 or more messages from a queue. The VT specifies the duration of time in seconds that the message is invisible to other consumers. At the end of that duration, the message is visible again and could be read by other consumers.

1
pgmq.read(
2
    queue_name text,
3
    vt integer,
4
    qty integer
5
)
6
7
returns setof pgmq.message_record

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`vt`	`integer`	消息被读取后变为不可见的时间（秒）
`qty`	`integer`	从队列中读取的消息数量，默认值为 1

示例：

🌐 Example:

1
select * from pgmq.read('my_queue', 10, 2);
2
 msg_id | read_ct |          enqueued_at          |              vt               |       message
3
--------+---------+-------------------------------+-------------------------------+----------------------
4
      1 |       1 | 2023-10-28 19:14:47.356595-05 | 2023-10-28 19:17:08.608922-05 | {"hello": "world_0"}
5
      2 |       1 | 2023-10-28 19:14:47.356595-05 | 2023-10-28 19:17:08.608974-05 | {"hello": "world_1"}
6
(2 rows)

`read_with_poll`#

和 read() 一样。还提供了方便的长轮询功能。当队列中没有消息时，函数调用会等待 max_poll_seconds 的时间后才返回。如果在这段时间内有消息到达队列，它们会被立即读取并返回。

🌐 Same as read(). Also provides convenient long-poll functionality. When there are no messages in the queue, the function call will wait for max_poll_seconds in duration before returning. If messages reach the queue during that duration, they will be read and returned immediately.

1
pgmq.read_with_poll(
2
    queue_name text,
3
    vt integer,
4
    qty integer,
5
    max_poll_seconds integer default 5,
6
    poll_interval_ms integer default 100
7
)
8
returns setof pgmq.message_record

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`vt`	`integer`	读取消息后消息变为不可见的时间（秒）
`qty`	`integer`	从队列中读取的消息数量。默认值为 1
`max_poll_seconds`	`integer`	等待新消息到达队列的时间（秒）。默认值为 5
`poll_interval_ms`	`integer`	内部轮询操作之间的间隔时间（毫秒）。默认值为 100

示例：

🌐 Example:

1
select * from pgmq.read_with_poll('my_queue', 1, 1, 5, 100);
2
 msg_id | read_ct |          enqueued_at          |              vt               |      message
3
--------+---------+-------------------------------+-------------------------------+--------------------
4
      1 |       1 | 2023-10-28 19:09:09.177756-05 | 2023-10-28 19:27:00.337929-05 | {"hello": "world"}

`pop`#

从队列中读取一条消息，并在读取后将其删除。

🌐 Reads a single message from a queue and deletes it upon read.

注意：如果消费应用不能保证消息被处理，使用 pop() 会导致消息最多只被投递一次。

🌐 Note: utilization of pop() results in at-most-once delivery semantics if the consuming application does not guarantee processing of the message.

1
pgmq.pop(queue_name text)
2
returns setof pgmq.message_record

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

1
pgmq=# select * from pgmq.pop('my_queue');
2
 msg_id | read_ct |          enqueued_at          |              vt               |      message
3
--------+---------+-------------------------------+-------------------------------+--------------------
4
      1 |       2 | 2023-10-28 19:09:09.177756-05 | 2023-10-28 19:27:00.337929-05 | {"hello": "world"}

删除/存档消息 #

🌐 Deleting/Archiving messages

`delete`#

🌐 delete (single)

从队列中删除一条消息。

🌐 Deletes a single message from a queue.

1
pgmq.delete (queue_name text, msg_id: bigint)
2
returns boolean

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg_id`	`bigint`	要删除的消息的消息ID

示例：

🌐 Example:

1
select pgmq.delete('my_queue', 5);
2
 delete
3
--------
4
 t

`delete`#

🌐 delete (batch)

从队列中删除一条或多条消息。

🌐 Delete one or many messages from a queue.

1
pgmq.delete (queue_name text, msg_ids: bigint[])
2
returns setof bigint

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg_ids`	`bigint[]`	要删除的消息 ID 数组

示例：

🌐 Examples:

删除两条存在的消息。

🌐 Delete two messages that exist.

1
select * from pgmq.delete('my_queue', array[2, 3]);
2
 delete
3
--------
4
      2
5
      3

删除两条消息，一条存在，一条不存在。消息 999 不存在。

🌐 Delete two messages, one that exists and one that does not. Message 999 does not exist.

1
select * from pgmq.delete('my_queue', array[6, 999]);
2
 delete
3
--------
4
      6

`purge_queue`#

永久删除队列中的所有消息。会返回已删除的消息数量。

🌐 Permanently deletes all messages in a queue. Returns the number of messages that were deleted.

1
purge_queue(queue_name text)
2
returns bigint

参数：

参数	类型	描述
queue_name	文本	队列的名称

示例：

🌐 Example:

当队列中有8条消息时清空它；

🌐 Purge the queue when it contains 8 messages;

1
select * from pgmq.purge_queue('my_queue');
2
 purge_queue
3
-------------
4
           8

`archive`#

🌐 archive (single)

从指定的队列中移除一条请求的消息，并将其插入队列的归档中。

🌐 Removes a single requested message from the specified queue and inserts it into the queue's archive.

1
pgmq.archive(queue_name text, msg_id bigint)
2
returns boolean

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg_id`	`bigint`	要归档的消息的消息ID

返回值布尔值，表示操作是成功还是失败。

🌐 Returns Boolean value indicating success or failure of the operation.

示例：从队列 my_queue 中删除 ID 为 1 的消息并将其归档：

🌐 Example; remove message with ID 1 from queue my_queue and archive it:

1
select * from pgmq.archive('my_queue', 1);
2
 archive
3
---------
4
       t

`archive`#

🌐 archive (batch)

从指定队列中删除一批请求的消息，并将它们插入到队列的归档中。返回一个已成功归档的消息 ID 数组。

🌐 Deletes a batch of requested messages from the specified queue and inserts them into the queue's archive. Returns an array of message ids that were successfully archived.

1
pgmq.archive(queue_name text, msg_ids bigint[])
2
RETURNS SETOF bigint

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg_ids`	`bigint[]`	要归档的消息ID数组

示例：

🌐 Examples:

删除队列 my_queue 中 ID 为 1 和 2 的消息，然后移动到存档。

🌐 Delete messages with ID 1 and 2 from queue my_queue and move to the archive.

1
select * from pgmq.archive('my_queue', array[1, 2]);
2
 archive
3
---------
4
       1
5
       2

删除消息4（存在的）和999（不存在的）。

🌐 Delete messages 4, which exists and 999, which does not exist.

1
select * from pgmq.archive('my_queue', array[4, 999]);
2
 archive
3
---------
4
       4

工具 #

🌐 Utilities

`set_vt`#

将消息的可见性超时设置为未来指定的时间段。返回已更新的消息记录。

🌐 Sets the visibility timeout of a message to a specified time duration in the future. Returns the record of the message that was updated.

1
pgmq.set_vt(
2
    queue_name text,
3
    msg_id bigint,
4
    vt_offset integer
5
)
6
returns pgmq.message_record

参数：

参数	类型	描述
`queue_name`	`text`	队列的名称
`msg_id`	`bigint`	要设置可见时间的消息 ID
`vt_offset`	`integer`	从现在开始，消息 VT 要设置的时间长度（秒）

示例：

🌐 Example:

把消息 1 的可见性超时设置为从现在起 30 秒。

🌐 Set the visibility timeout of message 1 to 30 seconds from now.

1
select * from pgmq.set_vt('my_queue', 11, 30);
2
 msg_id | read_ct |          enqueued_at          |              vt               |       message
3
--------+---------+-------------------------------+-------------------------------+----------------------
4
     1 |       0 | 2023-10-28 19:42:21.778741-05 | 2023-10-28 19:59:34.286462-05 | {"hello": "world_0"}

`list_queues`#

列出目前存在的所有队列。

🌐 List all the queues that currently exist.

1
list_queues()
2
RETURNS TABLE(
3
    queue_name text,
4
    created_at timestamp with time zone,
5
    is_partitioned boolean,
6
    is_unlogged boolean
7
)

示例：

🌐 Example:

1
select * from pgmq.list_queues();
2
      queue_name      |          created_at           | is_partitioned | is_unlogged
3
----------------------+-------------------------------+----------------+-------------
4
 my_queue             | 2023-10-28 14:13:17.092576-05 | f              | f
5
 my_partitioned_queue | 2023-10-28 19:47:37.098692-05 | t              | f
6
 my_unlogged          | 2023-10-28 20:02:30.976109-05 | f              | t

`metrics`#

获取特定队列的指标。

🌐 Get metrics for a specific queue.

1
pgmq.metrics(queue_name: text)
2
returns table(
3
    queue_name text,
4
    queue_length bigint,
5
    newest_msg_age_sec integer,
6
    oldest_msg_age_sec integer,
7
    total_messages bigint,
8
    scrape_time timestamp with time zone
9
)

参数：

参数	类型	描述
queue_name	文本	队列的名称

返回：

属性	类型	描述
`queue_name`	`text`	队列的名称
`queue_length`	`bigint`	当前队列中的消息数量
`newest_msg_age_sec`	`integer	null`	队列中新消息的存在时间，单位为秒
`oldest_msg_age_sec`	`integer	null`	队列中最旧消息的存在时间，单位为秒
`total_messages`	`bigint`	队列中累计通过的消息总数
`scrape_time`	`timestamp with time zone`	当前时间戳

示例：

🌐 Example:

1
select * from pgmq.metrics('my_queue');
2
 queue_name | queue_length | newest_msg_age_sec | oldest_msg_age_sec | total_messages |          scrape_time
3
------------+--------------+--------------------+--------------------+----------------+-------------------------------
4
 my_queue   |           16 |               2445 |               2447 |             35 | 2023-10-28 20:23:08.406259-05

`metrics_all`#

获取所有现有队列的指标。

🌐 Get metrics for all existing queues.

1
pgmq.metrics_all()
2
RETURNS TABLE(
3
    queue_name text,
4
    queue_length bigint,
5
    newest_msg_age_sec integer,
6
    oldest_msg_age_sec integer,
7
    total_messages bigint,
8
    scrape_time timestamp with time zone
9
)

返回：

属性	类型	描述
`queue_name`	`text`	队列的名称
`queue_length`	`bigint`	当前队列中的消息数量
`newest_msg_age_sec`	`integer	null`	队列中新消息的存在时间，单位为秒
`oldest_msg_age_sec`	`integer	null`	队列中最旧消息的存在时间，单位为秒
`total_messages`	`bigint`	队列中累计通过的消息总数
`scrape_time`	`timestamp with time zone`	当前时间戳

1
select * from pgmq.metrics_all();
2
      queue_name      | queue_length | newest_msg_age_sec | oldest_msg_age_sec | total_messages |          scrape_time
3
----------------------+--------------+--------------------+--------------------+----------------+-------------------------------
4
 my_queue             |           16 |               2563 |               2565 |             35 | 2023-10-28 20:25:07.016413-05
5
 my_partitioned_queue |            1 |                 11 |                 11 |              1 | 2023-10-28 20:25:07.016413-05
6
 my_unlogged          |            1 |                  3 |                  3 |              1 | 2023-10-28 20:25:07.016413-05

类型 #

🌐 Types

`message_record`#

队列中消息的完整表示。

🌐 The complete representation of a message in a queue.

属性名	类型	描述
`msg_id`	`bigint`	消息的唯一ID
`read_ct`	`bigint`	消息被阅读的次数。每次调用read()时增加
`enqueued_at`	`timestamp with time zone`	消息被插入队列的时间
`vt`	`timestamp with time zone`	消息对消费者可读的时间戳
`message`	`jsonb`	消息内容

示例：

🌐 Example:

1
msg_id | read_ct |          enqueued_at          |              vt               |      message
2
--------+---------+-------------------------------+-------------------------------+--------------------
3
      1 |       1 | 2023-10-28 19:06:19.941509-05 | 2023-10-28 19:06:27.419392-05 | {"hello": "world"}

资源 #

🌐 Resources

官方文档：pgmq/api

功能 #

启用扩展 #

使用情况 #

排队管理 #

create#

create_unlogged#

detach_archive#

drop_queue#

发送消息 #

send#

send_batch#

看消息 #

read#

read_with_poll#

pop#

删除/存档消息 #

delete#

delete#

purge_queue#

archive#

archive#

工具 #

set_vt#

list_queues#

metrics#

metrics_all#

类型 #

message_record#

资源 #

`create`#

`create_unlogged`#

`detach_archive`#

`drop_queue`#

`send`#

`send_batch`#

`read`#

`read_with_poll`#

`pop`#

`delete`#

`delete`#

`purge_queue`#

`archive`#

`archive`#

`set_vt`#

`list_queues`#

`metrics`#

`metrics_all`#

`message_record`#