Azure Database for PostgreSQL 灵活服务器中的逻辑复制和逻辑解码

Azure Database for PostgreSQL 灵活服务器支持以下逻辑数据提取和复制方法：

逻辑复制
1. 使用 PostgreSQL 本机逻辑复制复制数据对象。逻辑复制提供对数据复制的精细控制，包括表级数据复制。
2. 使用 pglogical 扩展，提供逻辑流复制以及更多功能，例如复制数据库的初始架构、支持 TRUNCATE、复制 DDL 等。
通过解码预写日志（WAL）的内容来实现的逻辑解码。

比较逻辑复制和逻辑解码

逻辑复制和逻辑解码具有一些相似之处。这两者都：

允许从 Postgres 复制数据。
使用预写日志 (WAL) 作为更改源。
使用逻辑复制槽发送数据。槽表示更改流。
使用表的 REPLICA IDENTITY 属性确定可发送出去的更改内容。
不要复制 DDL 更改。

这两种技术存在不同之处：

逻辑复制：

允许你指定要复制的一个表或一组表。

逻辑解码：

提取数据库中所有表的更改。

逻辑复制和逻辑解码的先决条件

转到门户上的“参数”页。
将参数 wal_level 设置为 logical。
如果要使用 pglogical 扩展，请搜索 shared_preload_libraries 和 azure.extensions 参数，然后从下拉列表框中选择 pglogical 。
将 max_worker_processes 参数值更新为至少 16。否则，可能会遇到类似 WARNING: out of background worker slots问题。
保存更改并重启服务器以应用更改。
确认Azure Database for PostgreSQL灵活服务器允许来自连接资源的网络流量。

授予管理员用户复制权限。

ALTER ROLE <adminname> WITH REPLICATION;

请确保所使用的角色对要复制的架构具有特权。否则，你可能会遇到诸如 Permission denied for schema 之类的错误。

注释

最好将复制用户与常规管理员帐户分开。

使用逻辑复制和逻辑解码

使用本机逻辑复制是将数据从Azure Database for PostgreSQL灵活服务器复制的最简单方法。你可以使用 SQL 接口或流式处理协议来使用更改。你还可以通过 SQL 接口使用逻辑解码来获取变更。

本机逻辑复制

逻辑复制使用术语 发布服务器 和 订阅服务器。

发布服务器是发送数据的Azure Database for PostgreSQL灵活服务器数据库。
订阅者是接收数据的 Azure Database for PostgreSQL 灵活服务器数据库。

下面是可用于尝试逻辑复制的一些示例代码。

连接到发布服务器数据库。创建表并添加一些数据。

CREATE TABLE basic (id INTEGER NOT NULL PRIMARY KEY, a TEXT);
INSERT INTO basic VALUES (1, 'apple');
INSERT INTO basic VALUES (2, 'banana');

为表创建发布。

CREATE PUBLICATION pub FOR TABLE basic;

连接到订阅服务器数据库。使用发布服务器上的同一架构创建表。
```
CREATE TABLE basic (id INTEGER NOT NULL PRIMARY KEY, a TEXT);
```

创建连接到之前创建的发布的订阅。

CREATE SUBSCRIPTION sub CONNECTION 'host=<server>.postgres.database.chinacloudapi.cn user=<rep_user> dbname=<dbname> password=<password>' PUBLICATION pub;

现在，可以在订阅服务器上查询表。你会看到它从发布者接收数据。
```
SELECT * FROM basic;
```
可将更多的行添加到发布服务器的表中，并查看订阅服务器上的更改。

如果看不到数据，请切换到属于 azure_pg_admin 角色的用户，并检查表中的内容。

若要详细了解逻辑复制，请访问 PostgreSQL 文档。

在同一服务器上的数据库之间使用逻辑复制

若要在同一Azure Database for PostgreSQL灵活服务器上设置不同数据库的逻辑复制，请遵循特定准则以避免实施限制。目前，如果复制槽未在同一命令中创建，则只能创建连接到同一数据库群集的订阅。否则，CREATE SUBSCRIPTION 调用会在 LibPQWalReceiverReceive 等待事件上挂起。此行为是由于 Postgres 引擎中的现有限制，可能会在将来的版本中将其删除。

若要在同一服务器上设置“源”和“目标”数据库之间的逻辑复制，同时避免此限制，请执行以下步骤：

首先，在源数据库和目标数据库中创建具有相同 basic 架构的表：

-- Run this on both source and target databases
CREATE TABLE basic (id INTEGER NOT NULL PRIMARY KEY, a TEXT);

接下来，在源数据库中，为该表创建发布，然后使用 pg_create_logical_replication_slot 函数单独创建一个逻辑复制槽。这种方法有助于防止通常在与订阅相同的命令中创建插槽时发生的挂起问题。使用 pgoutput 插件：

-- Run this on the source database
CREATE PUBLICATION pub FOR TABLE basic;
SELECT pg_create_logical_replication_slot('myslot', 'pgoutput');

然后，在目标数据库中，为先前创建的发布创建订阅。将 create_slot 设置为 false，以防止 Azure Database for PostgreSQL 灵活服务器创建新的槽位，并指定你在上一步中创建的槽位名称。在运行命令之前，请将连接字符串中的占位符替换为您的实际数据库凭据：

-- Run this on the target database
CREATE SUBSCRIPTION sub
   CONNECTION 'dbname=<source dbname> host=<server>.postgres.database.chinacloudapi.cn port=5432 user=<rep_user> password=<password>'
   PUBLICATION pub
   WITH (create_slot = false, slot_name='myslot');

设置逻辑复制后，通过将新记录插入源数据库中的 basic 表中，然后验证它是否复制到目标数据库来测试它：

-- Run this on the source database
INSERT INTO basic SELECT 3, 'mango';

-- Run this on the target database
TABLE basic;

如果已正确配置所有内容，则会在目标数据库中看到源数据库中的新记录，确认成功设置逻辑复制。

pglogical 扩展

下面是在提供者数据库服务器和订阅服务器上配置 pglogical 的示例。有关详细信息，请参阅 pglogical 扩展文档。此外，请确保完成前面列出的先决条件任务。

在发布端和订阅端数据库服务器上的数据库中安装 pglogical 扩展。
```
\c myDB
CREATE EXTENSION pglogical;
```
如果复制用户不是服务器管理用户（创建服务器的用户），请授予角色中的 azure_pg_admin 用户成员身份，并将 REPLICATION 和 LOGIN 属性分配给用户。有关详细信息，请参阅 pglogical 文档。
```
GRANT azure_pg_admin to myUser;
ALTER ROLE myUser REPLICATION LOGIN;
```

在提供者（源/发布者）数据库服务器上，创建提供者节点。

select pglogical.create_node( node_name := 'provider1',
dsn := ' host=myProviderServer.postgres.database.chinacloudapi.cn port=5432 dbname=myDB user=myUser password=<password>');

创建一个复制集。

select pglogical.create_replication_set('myreplicationset');

将数据库中的所有表添加到该复制集。

SELECT pglogical.replication_set_add_all_tables('myreplicationset', '{public}'::text[]);

也可将特定架构（例如 testUser）中的表添加到默认的复制集，这是替代方法。

SELECT pglogical.replication_set_add_all_tables('default', ARRAY['testUser']);

在订阅者数据库服务器上，创建一个订阅者节点。

select pglogical.create_node( node_name := 'subscriber1',
dsn := ' host=mySubscriberServer.postgres.database.chinacloudapi.cn port=5432 dbname=myDB user=myUser password=<password>' );

创建一个订阅以开始同步和复制过程。

select pglogical.create_subscription (
subscription_name := 'subscription1',
replication_sets := array['myreplicationset'],
provider_dsn := 'host=myProviderServer.postgres.database.chinacloudapi.cn port=5432 dbname=myDB user=myUser password=<password>');

验证订阅状态。

SELECT subscription_name, status FROM pglogical.show_subscription_status();

注意

Pglogical 目前不支持自动 DDL 复制。您可以使用 pg_dump --schema-only 手动复制初始架构。可以使用函数 pglogical.replicate_ddl_command 同时在提供程序和订阅服务器上执行 DDL 语句。请注意此处列出的扩展的其他限制。

逻辑解码

可以通过流式协议或 SQL 接口获取逻辑解码数据。

流式处理协议

使用流式协议来消费变更通常更可取。可以创建自己的使用者或连接器，或使用 Debezium 等第三方服务。

有关使用流式处理协议 pg_recvlogical的示例，请参阅 wal2json 文档：将流式处理协议与 pg_recvlogical 配合使用的示例。

SQL 接口

在以下示例中，将 SQL 接口与 wal2json 插件配合使用。

创建槽。

SELECT * FROM pg_create_logical_replication_slot('test_slot', 'wal2json');

发出 SQL 命令。例如：

CREATE TABLE a_table (
   id varchar(40) NOT NULL,
   item varchar(40),
   PRIMARY KEY (id)
);

INSERT INTO a_table (id, item) VALUES ('id1', 'item1');
DELETE FROM a_table WHERE id='id1';

使用更改。

SELECT data FROM pg_logical_slot_get_changes('test_slot', NULL, NULL, 'pretty-print', '1');

输出如下所示：

{
      "change": [
      ]
}
{
      "change": [
               {
                        "kind": "insert",
                        "schema": "public",
                        "table": "a_table",
                        "columnnames": ["id", "item"],
                        "columntypes": ["character varying(40)", "character varying(40)"],
                        "columnvalues": ["id1", "item1"]
               }
      ]
}
{
      "change": [
               {
                        "kind": "delete",
                        "schema": "public",
                        "table": "a_table",
                        "oldkeys": {
                              "keynames": ["id"],
                              "keytypes": ["character varying(40)"],
                              "keyvalues": ["id1"]
                        }
               }
      ]
}

使用完毕后，请删除该槽位。

SELECT pg_drop_replication_slot('test_slot');

若要了解有关逻辑解码的详细信息，请参阅 PostgreSQL 文档：逻辑解码。

Monitor

必须监视逻辑解码。删除任何未使用的复制槽。在这些更改被读取之前，槽位会一直保留 Postgres WAL 日志和相关系统目录。如果你的订阅服务器或使用者失败或者配置不正确，则未使用的日志将不断堆积，直至填满存储。此外，未使用的日志会增大事务 ID 换行的风险。这两种情况可能会导致服务器不可用。因此，必须持续使用逻辑复制槽。如果不再使用某个逻辑复制槽，请立即将其删除。

active视图中的pg_replication_slots列指示是否有消费者连接到插槽。

SELECT * FROM pg_replication_slots;

设置“已用最大事务 ID”和“已用存储空间”指标的警报，以在值超过正常阈值时通知你。

局限性

逻辑复制限制适用，如此处所述。
复制槽和 HA 故障转移 - 在 PostgreSQL 16 及更早版本中，将启用了高可用性 (HA) 的服务器与 Azure Database for PostgreSQL 一起使用时，发生故障转移时不会保留逻辑复制槽。若要维护逻辑复制槽位并确保故障转移后的数据一致性，请使用 PG 故障转移槽扩展并配置支持设置，例如 hot_standby_feedback = on。有关启用此扩展的详细信息，请参阅文档。

对逻辑复制槽的故障转移支持

对于 PostgreSQL 17 及更高版本，已原生支持复制槽同步。如果启用正确的 PostgreSQL 配置（sync_replication_slots，hot_standby_feedback），则故障转移后逻辑复制槽会自动保留，无需扩展。

重要

如果相应的订阅服务器不再存在，则必须删除主服务器中的逻辑复制槽。否则，WAL 文件开始在主存储中累积，从而填满存储。当存储使用率达到 95% 或可用容量小于 5 GiB 时，主服务器会自动切换到只读模式。如果存储阈值超过某个限制，并且逻辑复制槽未被使用（由于订阅服务器不可用），Azure Database for PostgreSQL 灵活服务器会自动删除该未使用的逻辑复制槽。此操作会释放累积的 WAL 文件，避免服务器因存储空间被占满而不可用。

Azure Database for PostgreSQL 中的 Firewall 规则。
具有公共访问权限的Azure Database for PostgreSQL的Networking 概述。
Azure Database for PostgreSQL 中的虚拟网络集成。
如何使用扩展。
Azure Database for PostgreSQL 中的高可用性。

Last updated on 2026-07-24