python - 使用 psycopg2 和publishers进行逻辑复制

Question

有用 python 和 psycopg2 编写的逻辑复制脚本: https://www.psycopg.org/docs/advanced.html#logical-replication-quick-start

它有效。但我不想接收数据库中的所有操作，而只想接收发布者中指定的操作。

为此:

1. 我在源数据库(版本 13.3)中创建了名为“PUB”的发布者和一个表。我还部署了第二个数据库(版本 14.1)并在其中创建了对此发布的订阅，以检查该发布的功能。并且它有效。

2. 在 psycopg2 文档 ( https://www.psycopg.org/docs/extras.html#replication-support-objects ) 和 postgresql 文档 ( https://postgrespro.ru/docs/postgresql/13/protocol-logical-replication?lang=en ) 中，我找到了一种在 start_replication 中设置publication_names 的方法

cur.start_replication(slot_name='pytest', decode=True, options={'publication_names': 'PUB'})

但是没有成功。我收到错误:

psycopg2.errors.InvalidParameterValue: option "publication_names" = "PUB," is unknown

其实就是这个问题。 psycorpg2 中的逻辑复制可以与订阅关联吗？如果是这样，怎么办？

P.s.有替代方案(触发/通知 -> 监听/psycopg2)，但如果可能的话我想解决这个问题。

Answer 1

您的问题的答案可能是“是和否”。

如jjanes状态在answer ，您必须使用 pgoutput 输出插件才能使用出版物。您可以配置 psycopg2 来使用它，如下所示:

conn = psycopg2.connect(
    dbname='test',
    connection_factory=psycopg2.extras.LogicalReplicationConnection,
)

options = {'publication_names': 'pub', 'proto_version': '1'}
cur = conn.cursor()
try:
    cur.start_replication(slot_name='pytest', decode=False, options=options)
except psycopg2.ProgrammingError:
    cur.create_replication_slot(
        'pytest',
        output_plugin='pgoutput',
    )
    cur.start_replication(slot_name='pytest', decode=False, options=options)

但是pgoutput使用二进制协议(protocol)，因此这些语句的输出

insert into table1 (col) values ('hello hello');
update table1 set col = 'hello goodbye' where id = 6;

是

b'B\x00\x00\x00\x00r!<x\x00\x02t\xc65\xbb\xb5\xd3\x00\x00jd'
b'R\x00\x00F:public\x00table1\x00d\x00\x02\x01id\x00\x00\x00\x00\x17\xff\xff\xff\xff\x00col\x00\x00\x00\x04\x13\x00\x00\x00\x14'
b'I\x00\x00F:N\x00\x02t\x00\x00\x00\x016t\x00\x00\x00\x0bhello hello'
b'C\x00\x00\x00\x00\x00r!<x\x00\x00\x00\x00r!<\xa8\x00\x02t\xc65\xbb\xb5\xd3'
b'B\x00\x00\x00\x00r!=8\x00\x02t\xc67`\xb5\xae\x00\x00je'
b'U\x00\x00F:N\x00\x02t\x00\x00\x00\x016t\x00\x00\x00\rhello goodbye'
b'C\x00\x00\x00\x00\x00r!=8\x00\x00\x00\x00r!=h\x00\x02t\xc67`\xb5\xae'

因为psycopg2不知道如何解码消息。您可以使用 struct 自行解码它们Python 标准库中的模块和 published message formats ，或者您可以考虑使用 wal2json而不是 pgoutput。 wal2json 目前不支持发布，但它确实提供了 emulate 的方法。他们。