跳转到主要内容

pgq

通用队列的PG实现

概览

扩展包名版本分类许可证语言
pgq3.5.2FEATISCC
ID扩展名BinLibLoadCreateTrustReloc模式
2650pgq否是否是否否pg_catalog

版本

类型仓库版本PG 大版本包名依赖
EXTPGDG3.5.21817161514pgq-
RPMPGDG3.5.11817161514pgq_$v-
DEBPGDG3.5.11817161514postgresql-$v-pgq3-
OS / PGPG18PG17PG16PG15PG14
el8.x86_64
el8.aarch64
el9.x86_64
el9.aarch64
el10.x86_64
el10.aarch64
d12.x86_64
d12.aarch64
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
d13.x86_64
d13.aarch64
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
PGDG 3.5.1
u22.x86_64
u22.aarch64
u24.x86_64
u24.aarch64
u26.x86_64
u26.aarch64

安装

您可以直接安装 pgq 扩展包的预置二进制包,首先确保 PGDG 仓库已经添加并启用:

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

使用 pig 或者是 apt/yum/dnf 安装扩展:

安装
pig install pgq;          # 当前活跃 PG 版本安装
pig
pig ext install -y pgq -v 18  # PG 18
pig ext install -y pgq -v 17  # PG 17
pig ext install -y pgq -v 16  # PG 16
pig ext install -y pgq -v 15  # PG 15
pig ext install -y pgq -v 14  # PG 14
dnf
dnf install -y pgq_18       # PG 18
dnf install -y pgq_17       # PG 17
dnf install -y pgq_16       # PG 16
dnf install -y pgq_15       # PG 15
dnf install -y pgq_14       # PG 14
apt
apt install -y postgresql-18-pgq3   # PG 18
apt install -y postgresql-17-pgq3   # PG 17
apt install -y postgresql-16-pgq3   # PG 16
apt install -y postgresql-15-pgq3   # PG 15
apt install -y postgresql-14-pgq3   # PG 14

创建扩展:

CREATE EXTENSION pgq;

用法

来源:

PgQ 是一个 PostgreSQL 扩展,提供通用的高性能无锁队列,带有简单的 SQL 函数 API。它使用生产者-消费者模型,基于批次进行事件处理。

CREATE EXTENSION pgq;

核心概念

  • 队列(Queue):命名的事件流。生产者插入事件,消费者按批次消费。
  • 消费者(Consumer):注册到队列上的命名订阅者。每个消费者跟踪自己的位置。
  • 批次(Batch):一组一起获取的事件。消费者逐批处理事件。
  • 心跳进程(Ticker):后台进程,定期创建批次边界(tick)。

队列管理

-- Create a queue
SELECT pgq.create_queue('myqueue');

-- Drop a queue
SELECT pgq.drop_queue('myqueue');

-- Get queue info
SELECT * FROM pgq.get_queue_info();
SELECT * FROM pgq.get_queue_info('myqueue');

消费者注册

-- Register a consumer on a queue
SELECT pgq.register_consumer('myqueue', 'myconsumer');

-- Unregister a consumer
SELECT pgq.unregister_consumer('myqueue', 'myconsumer');

-- Get consumer info
SELECT * FROM pgq.get_consumer_info('myqueue');

生产事件

-- Insert an event into a queue
SELECT pgq.insert_event('myqueue', 'event_type', 'event_data');

-- Insert with extra fields
SELECT pgq.insert_event('myqueue', 'event_type', 'event_data',
                         'extra1', 'extra2', 'extra3', 'extra4');

消费事件

-- Get the next batch of events (returns batch_id or NULL if no new batches)
SELECT pgq.next_batch('myqueue', 'myconsumer');

-- Get events from the batch
SELECT * FROM pgq.get_batch_events(:batch_id);

-- Retry a failed event (will reappear after the specified interval)
SELECT pgq.event_retry(:batch_id, :event_id, :retry_seconds);

-- Mark batch as done
SELECT pgq.finish_batch(:batch_id);

典型消费者循环

-- 1. Get next batch
SELECT pgq.next_batch('myqueue', 'myconsumer') AS batch_id;

-- 2. If batch_id is not NULL, get events
SELECT * FROM pgq.get_batch_events(:batch_id);

-- 3. Process events, retry failures
SELECT pgq.event_retry(:batch_id, :event_id, 60);

-- 4. Finish the batch
SELECT pgq.finish_batch(:batch_id);

维护

PgQ 需要在后台运行心跳守护进程(pgqd),用于创建批次边界并执行表轮转和重试事件处理等维护任务。

主要函数

函数描述
pgq.create_queue(name)创建新队列
pgq.drop_queue(name)删除队列
pgq.register_consumer(queue, consumer)注册消费者
pgq.unregister_consumer(queue, consumer)注销消费者
pgq.insert_event(queue, type, data, ...)插入事件
pgq.next_batch(queue, consumer)获取下一批次 ID
pgq.get_batch_events(batch_id)从批次获取事件
pgq.event_retry(batch_id, event_id, seconds)安排事件重试
pgq.finish_batch(batch_id)标记批次已处理
pgq.get_queue_info([name])获取队列统计信息
pgq.get_consumer_info(queue)获取消费者统计信息

版本与访问边界

上游 3.5.2 支持 PostgreSQL 10 至 19;3.5.2 新增 PostgreSQL 19 支持,未记载 SQL API 变化。目录当前记录的软件包版本为 3.5.1,选择扩展升级目标前须确认已安装文件。控制文件要求由超级用户安装,且不允许迁移模式。API 使用 pgq 模式;应逐一检查应用角色的队列管理与消费权限。

只有 pgq.finish_batch 成功后,消费者位置才会前进。尽可能让事件处理与确认在同一事务内完成;外部副作用无法加入数据库事务时,应处理重复投递。

这个页面对您有帮助吗?