pgqueue 0.6

PgQ介绍

(来自SkyTools README)

PgQ是一个用PL/pgSQL、Python和C代码编写的排队系统。它基于Slony-I的基于快照的事件处理思想，适用于通用用途。

PgQ提供了一个高效、事务性的排队系统，支持多节点（包括工作共享和分割、故障转移和切换，适用于队列和消费者）。

规则

• 数据库中可以有多个队列。

• 可以有多个生产者可以插入任何队列。

• 一个队列可以有多个消费者。

• 消费者可以有多个子消费者。

PgQ分为3层：生产者、计时器和消费者。

生产者和消费者分别向队列中推送和读取事件。生产者只需要调用PostgreSQL存储过程（如表上的触发器或从应用程序中的PostgreSQL调用）。消费者通常用Python编写，但任何能够运行PostgreSQL存储过程的语言都可以使用。

Tick器是一个守护进程，它将队列分割成事件批次并处理系统的维护。

PgQueue 模块

该模块提供了用于编写 生产者 和 消费者 的 Python 函数和类。它还包含了对 SkyTools 中原始 C 语言 Tick器 的 Python 实现，它模仿了 SkyTools 中的原始 C 语言 Tick器：分割事件批次并执行维护任务。

安装

先决条件

• Python >= 2.6 或 Python 3

• psycopg2 作为依赖项自动安装

• (在服务器上) PgQ 扩展版本 >= 3.1

在 Debian / Ubuntu 上，您需要添加 PostgreSQL APT 仓库，然后根据 PostgreSQL 版本安装包 postgresql-x.x-pgq3。

最后在数据库中创建扩展

CREATE EXTENSION IF NOT EXISTS pgq;

您可以将 pgqueue 模块安装到您的环境中。

pip install --update pgqueue

示例用法

您需要永久运行 Tick器。如果 Tick器关闭，事件将被存储到队列中，但不会为消费者准备批次，事件表将快速增长。

对于 Tick器，您可以选择优化后的 C 语言编写的 pgqd 多数据库 Tick器，它是 SkyTools 的一部分，或者使用此模块提供的更简单的 Python 实现。

python -m pgqueue 'host=127.0.0.1 port=5432 user=jules password=xxxx dbname=test_db'

让我们创建一个新的队列，并注册一个消费者

conn = psycopg2.connect("dbname=test user=postgres")
conn.autocommit = True
cursor = conn.cursor()

first_q = pgqueue.Queue('first_queue')
first_q.create(cursor, ticker_max_lag='4 seconds')

consum_q = pgqueue.Consumer('first_queue', 'consumer_one')
consum_q.register(cursor)

我们已经准备好将事件生产到队列中，并在应用程序中稍后消费事件

first_q.insert_event(cursor, 'announce', 'Hello ...')
first_q.insert_event(cursor, 'announce', 'Hello world!')

# ... wait a little bit

conn.autocommit = False
for event in consum_q.next_events(cursor, commit=True):
    print(event)

您可以通过浏览源代码了解高级用法，直到我们编写更多文档（欢迎贡献）。

有关更多详细信息，请参阅 上游文档。

致谢

PgQ 是由 Marko Kreen 开发的 PostgreSQL 扩展。它是 SkyTools 的一部分，SkyTools 是 Skype 中用于复制和故障转移的工具包。

SkyTools 还嵌入了一个 pgq Python 框架，它提供了一点点不同的 API。

链接

• 在 GitHub 上 Fork 我