# 使用Replication Slot创建数据订阅 #### 功能介绍 PostgreSQL原生支持使用复制槽（Replication Slot）开启数据订阅（Change Data Capture，简称CDC）。Replication Slot是用于逻辑复制的核心机制，主要作用如下： 1. 确保主库的WAL日志（预写日志）不会被过早清理，直到从库或订阅者消费完日志。 2. 记录订阅者的消费位置（LSN），避免数据丢失或重复消费。 3. 支持通过逻辑解码插件（例如test_decoding和wal2json）将WAL日志解析为可读的变更数据（例如INSERT、UPDATE、DELETE操作），实现数据订阅。 本文介绍在RDS for PostgreSQL实例上开启数据订阅的配置方法。 #### 前提条件 - 参考[自定义购买ECS](https://support.huaweicloud.com/usermanual-ecs/ecs_03_7002.html)，已购买ECS。ECS选择与RDS for PostgreSQL实例相同的区域、VPC和安全组，便于RDS for PostgreSQL和ECS网络互通。 - 参考[购买并通过PostgreSQL客户端连接RDS for PostgreSQL实例](https://support.huaweicloud.com/qs-rds-pg/rds_02_0016.html)，已购买RDS for PostgreSQL实例并安装PostgreSQL客户端。 - 参考[通过内网连接RDS for PostgreSQL实例（Linux方式）](https://support.huaweicloud.com/usermanual-rds-pg/rds_pg_connect_05.html#section1)，已通过ECS连接到RDS for PostgreSQL实例。 #### 注意事项 - 仅支持在主实例上开启数据订阅并消费，不支持只读实例。 - RDS for PostgreSQL支持Logical Replication Slot Failover，主备切换操作不会影响数据订阅。更多介绍，请参见[逻辑订阅故障转移（Failover Slot）](https://support.huaweicloud.com/kerneldesc-rds-pg/rds_12_0000.html)。 - 开启数据订阅前需要修改RDS for PostgreSQL实例参数，并重启RDS for PostgreSQL实例。请在业务低峰期进行修改，避免对业务造成影响。 #### 开启数据订阅 #### 步骤一：创建测试数据库 本示例以创建数据库**testdb**为例。 1. [登录RDS管理控制台](https://console.huaweicloud.com/rds/#/rds/management/list)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-rds-pg/zh-cn_image_0000002577997338.png)，选择区域。 3. 在"实例管理"页面，选择目标实例，单击实例名称，进入实例的"概览"页。 4. 在左侧导航栏，单击"数据库管理"，单击"创建数据库"，在弹出框中填写数据库信息，单击"确定"。 图1创建数据库 ![](https://support.huaweicloud.com/bestpractice-rds-pg/zh-cn_image_0000002441297725.png "点击放大") #### 步骤二：创建测试账号并配置权限 本示例创建的账号名仅为示例，您可以根据实际情况自定义名称。 1. 在账号管理页面，单击"创建账号"。 2. 在弹出框中，输入账号名称、密码、权限和备注，单击"确定"。 3. 创建数据订阅普通账号，例如**cdc_user**。 4. 使用**root** 账号连接RDS for PostgreSQL实例。 ``` psql -h <实例连接地址> -p 5432 -U root -d testdb ``` 5. 执行如下命令，将**cdc_user** 账号添加到Replication角色中，并查询修改结果。 ``` ALTER USER cdc_user WITH REPLICATION; SELECT rolreplication FROM pg_roles WHERE rolname='cdc_user'; ``` 查询结果： ``` rolreplication ---------------- t (1 row) ``` 6. 执行如下命令为**cdc_user** 账号授权。 ``` GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user; ``` #### 步骤三：调整RDS for PostgreSQL实例参数 1. 执行如下命令，查询实例参数设置。 ``` SELECT name, setting, short_desc, source FROM pg_settings WHERE name ='wal_level'; ``` 查询结果： ``` name | setting | short_desc | source -----------------------+---------+-------------------------------------------------------------------------+-------------------- wal_level | replica | Sets the level of information written to the WAL. | configuration file (1 rows) ``` wal_level参数控制WAL日志中记录的信息详细程度，其取值决定了WAL日志的用途（例如崩溃恢复、物理复制、逻辑复制等）。wal_level有如下三个主要取值，从低到高记录的信息依次增加： - minimal：仅记录崩溃恢复必需的最基础信息，不支持任何复制（几乎不用于生产环境）。 - replica（默认值）：记录物理复制所需的完整信息，支持物理复制（例如主备同步，备实例通过重做WAL日志与主实例保持物理存储一致）和只读副本（Hot Standby）。 - logical：在replica级别的基础上，额外记录逻辑复制所需的元数据（例如行级变更的具体内容、表结构信息等），支持逻辑复制和变更数据捕获（CDC）。 2. 在实例管理页面，单击实例名称，进入实例的"概览"页。 3. 在左侧导航栏，选择"参数设置"**。** 4. 将wal_level参数取值修改为logical。 修改实例参数的具体方法，请参见[修改RDS for PostgreSQL实例参数](https://support.huaweicloud.com/usermanual-rds-pg/rds_pg_configuration.html)。 请在业务低峰期修改实例参数，避免对业务造成影响。 5. 修改实例参数并提交后，您需要重启RDS for PostgreSQL实例。 #### 步骤四：创建Logical Replication Slot [步骤三：调整RDS for PostgreSQL实例参数]中修改了实例参数，请在RDS for PostgreSQL实例重启完成后，实例状态变为"正常"时再进行以下操作。 1. 使用**root** 账号连接RDS for PostgreSQL实例。 ``` psql -h <实例连接地址> -p 5432 -U root -d testdb ``` 2. 执行如下命令，指定解码插件为test_decoding，创建一个名为cdc_replication_slot的Replication Slot。 ``` SELECT * FROM pg_create_logical_replication_slot( 'cdc_replication_slot', 'test_decoding' ); ``` 执行成功后，返回结果包含复制槽名称和初始LSN（日志序列号），如下所示： ``` slot_name | lsn ----------------+----------- cdc_replication_slot | 0/16A00000 (1 row) ``` 3. 执行以下命令查看所有复制槽，验证复制槽创建成功。 ``` SELECT slot_name, plugin, active FROM pg_replication_slots; ``` 若slot_name为cdc_replication_slot且plugin为test_decoding，表示复制槽创建成功。 ``` slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size | two_phase ----------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+----------------+----------- cdc_replication_slot | test_decoding | logical | 31721 | testdb | f | f | | | 25523| 2/2500A1B0 | 2/25003882 | reserved | | f (1 row) ``` #### 步骤五：创建测试数据 执行以下命令，创建测试数据，模拟生产环境。 1. 创建测试表，用于后续生成变更数据。 ``` CREATE TABLE public.tb_tests ( id SERIAL PRIMARY KEY, product_name VARCHAR(50) NOT NULL, price NUMERIC(10,2) NOT NULL, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); ``` 2. 插入数据。 ``` INSERT INTO public.tb_tests (product_name, price) VALUES ('笔记本电脑', 5999.99); INSERT INTO public.tb_tests (product_name, price) VALUES ('机械键盘', 299.99); ``` 3. 更新数据。 ``` UPDATE public.tb_tests SET price = 5899.99 WHERE id = 1; ``` 4. 删除数据。 ``` DELETE FROM public.tb_tests WHERE id = 2; ``` #### 步骤六：客户端读取数据 1. 使用cdc_user账号连接RDS for PostgreSQL实例。 ``` psql -h <实例连接地址> -p 5432 -U cdc_user -d testdb ``` 2. 执行以下命令，从Replication Slot中读取数据。 ``` SELECT * FROM pg_logical_slot_peek_changes( 'cdc_replication_slot', -- 复制槽名称 NULL, -- 起始LSN（NULL表示从当前位置开始） NULL, -- 最大返回条目数（NULL表示无限制） ); ``` 查询结果示例： ``` lsn | xid | data ------------+-------+------------------------------------------------------------------------- 0/16A01200 | 1234 | BEGIN 1234 0/16A01200 | 1234 | table public.tb_tests: INSERT: id[integer]:1 product_name[character varying]:'笔记本电脑' price[numeric(10,2)]:5999.99 create_time[timestamp with time zone]:'2024-05-20 10:00:00+08' 0/16A01350 | 1234 | table public.tb_tests: INSERT: id[integer]:2 product_name[character varying]:'机械键盘' price[numeric(10,2)]:299.99 create_time[timestamp with time zone]:'2024-05-20 10:01:00+08' 0/16A014A0 | 1234 | COMMIT 1234 0/16A015F0 | 1235 | BEGIN 1235 0/16A015F0 | 1235 | table public.tb_tests: UPDATE: id[integer]:1 price[numeric(10,2)]:5899.99 (old:5999.99) 0/16A016E0 | 1235 | COMMIT 1235 0/16A017D0 | 1236 | BEGIN 1236 0/16A017D0 | 1236 | table public.tb_tests: DELETE: id[integer]:2 0/16A01860 | 1236 | COMMIT 1236 (8 rows) ``` #### 步骤七：客户端消费订阅数据 在客户端终端执行以下命令，连接数据库并从复制槽cdc_replication_slot消费数据。 ``` pg_recvlogical \ -h \ -p 5432 \ -U cdc_user \ -d testdb \ --slot=cdc_replication_slot \ --start \ -f - ``` 参数说明如下，请替换为实际值。 - -h：RDS for PostgreSQL主机地址 - -p：端口号，默认5432 - -U：订阅用户，需要具备REPLICATION权限。 - -d：数据库名 - --slot：复制槽名称 - --start：从当前位置开始消费，首次消费可省略，默认从起始位置。 - -f：输出到控制台，- 表示标准输出。 结果示例： ``` BEGIN 1234 table public.tb_tests: INSERT: id[integer]:1 product_name[character varying]:'笔记本电脑' price[numeric(10,2)]:5999.99 create_time[timestamp with time zone]:'2024-05-20 10:00:00+08' table public.tb_tests: INSERT: id[integer]:2 product_name[character varying]:'机械键盘' price[numeric(10,2)]:299.99 create_time[timestamp with time zone]:'2024-05-20 10:01:00+08' COMMIT 1234 BEGIN 1235 table public.tb_tests: UPDATE: id[integer]:1 price[numeric(10,2)]:5899.99 (old:5999.99) COMMIT 1235 BEGIN 1236 table public.tb_tests: DELETE: id[integer]:2 COMMIT 1236 ``` ![](https://support.huaweicloud.com/bestpractice-rds-pg/public_sys-resources/note_3.0-zh-cn.png) - 消费特性：pg_recvlogical消费后，数据会被标记为"已处理"，复制槽的restart_lsn会更新到最新位置，后续无论是pg_recvlogical还是pg_logical_slot_peek_changes都不会再返回这些数据（避免重复消费）。 - 持续监听：命令会保持连接状态，若后续有新的变更（例如再次执行INSERT或UPDATE），会实时输出新的事务数据。 - 终止消费：按Ctrl+C可终止命令，下次执行时会从上次终止的位置继续消费（依赖复制槽记录的restart_lsn）。 #### 关闭数据订阅 开启了数据订阅的RDS for PostgreSQL实例，需要使用更多的WAL日志存储空间，当不再需要数据订阅时，需通过以下步骤彻底关闭，避免复制槽持续占用WAL日志空间： 1. 使用**root** 账号连接RDS for PostgreSQL实例。 ``` psql -h <实例连接地址> -p 5432 -U root -d testdb ``` 2. （可选）确认复制槽存在。 ``` SELECT slot_name FROM pg_replication_slots WHERE slot_name = 'cdc_replication_slot'; ``` 3. 删除复制槽。该操作不可逆，请谨慎操作。 ``` SELECT pg_drop_replication_slot('cdc_replication_slot'); ``` 4. 验证数据订阅已关闭。 ``` SELECT slot_name FROM pg_replication_slots; ``` 如果查询结果中没有cdc_replication_slot，表示已关闭。