使用Replication Slot创建数据订阅

本示例以创建数据库testdb为例。

1. 登录管理控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击页面左上角的，选择“数据库 > 云数据库 RDS”，进入RDS信息页面。
4. 在“实例管理”页面，选择目标实例，单击实例名称，进入实例的“概览”页。
5. 在左侧导航栏，单击“数据库管理”，单击“创建数据库”，在弹出框中填写数据库信息，单击“确定”。
   图1 创建数据库
   

本示例创建的账号名仅为示例，您可以根据实际情况自定义名称。

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;

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实例参数。

   请在业务低峰期修改实例参数，避免对业务造成影响。

5. 修改实例参数并提交后，您需要重启RDS for PostgreSQL实例。

步骤三：调整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中读取数据。（peek表示仅查看，不标记为已消费）

   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 <PostgreSQL主机地址> \  
  -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

• 消费特性：pg_recvlogical消费后，数据会被标记为“已处理”，复制槽的restart_lsn会更新到最新位置，后续无论是pg_recvlogical还是pg_logical_slot_peek_changes都不会再返回这些数据（避免重复消费）。
• 持续监听：命令会保持连接状态，若后续有新的变更（例如再次执行INSERT或UPDATE），会实时输出新的事务数据。
• 终止消费：按Ctrl+C可终止命令，下次执行时会从上次终止的位置继续消费（依赖复制槽记录的restart_lsn）。