为 Amazon RDS for PostgreSQL 执行逻辑复制

为 RDS for PostgreSQL 数据库实例启用逻辑解码

1. 确保您使用的用户账户具有以下角色：

   • rds_superuser 角色，以使您可以启用逻辑复制

   • rds_replication 角色，以授予管理逻辑槽并使用逻辑槽流式处理数据的权限

2. 将 rds.logical_replication 静态参数设置为 1。在应用该参数时，还将设置参数 wal_level、max_wal_senders、max_replication_slots 和 max_connections。这些参数更改可能会增加 WAL 生成，因此，仅在使用逻辑槽时设置 rds.logical_replication 参数。

3. 重启数据库实例，静态 rds.logical_replication 参数才会生效。

4. 请按下一部分中的说明创建逻辑复制槽。该过程需要您指定解码插件。目前，RDS for PostgreSQL 支持 PostgreSQL 随附的 test_decoding 和 wal2json 输出插件。

1. 创建源表

   连接到 RDS for PostgreSQL 数据库实例中的源数据库。

   
   source=> CREATE TABLE testtab (slno int primary key);
   CREATE TABLE
           

2. 将数据插入到源表中

   
   source=> INSERT INTO testtab VALUES (generate_series(1,1000));
   INSERT 0 1000
           

3. 为源表创建发布

   • 为源表创建发布：

     
     source=> CREATE PUBLICATION testpub FOR TABLE testtab;
     CREATE PUBLICATION
             

   • 使用 SELECT 查询来验证已创建的发布的详细信息：

     
     source=> SELECT * FROM pg_publication;
       oid   | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot
     --------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------
      115069 | testpub |    16395 | f            | t         | t         | t         | t           | f
     (1 row)
             

   • 验证源表已添加到发布中：

     
     source=> SELECT * FROM pg_publication_tables; 
     pubname | schemaname | tablename
     ---------+------------+-----------
      testpub | public     | testtab
     (1 rows)
             

   • 要复制数据库中的所有表，请使用：

     
     CREATE PUBLICATION testpub FOR ALL TABLES;
               

   • 如果已经为单个表创建了发布，并且您需要添加新表，则可以运行下面的查询来将任何新表添加到现有发布中：

     
     ALTER PUBLICATION <publication_name> add table <new_table_name>;
               

4. 连接到目标数据库并创建目标表

   • 连接到目标数据库实例中的目标数据库。创建与源表同名的目标表：

     
     target=> CREATE TABLE testtab (slno int primary key);
     CREATE TABLE
     

   • 通过对目标表运行 SELECT 查询，确保目标表中不存在数据：

         
     target=> SELECT count(*) FROM testtab;
      count
     -------
          0
     (1 row)
             

5. 在目标数据库中创建并验证订阅

   • 在目标数据库中创建订阅：

     
     target=> CREATE SUBSCRIPTION testsub 
     CONNECTION 'host=<source RDS/host endpoint> port=5432 dbname=<source_db_name> user=<user> password=<password>' 
     PUBLICATION testpub;
     NOTICE:  Created replication slot "testsub" on publisher
     CREATE SUBSCRIPTION
             

   • 使用 SELECT 查询来验证是否启用了订阅：

     
     target=> SELECT oid, subname, subenabled, subslotname, subpublications FROM pg_subscription;
       oid  | subname | subenabled | subslotname | subpublications
     -------+---------+------------+-------------+-----------------
      16434 | testsub | t          | testsub     | {testpub}
     (1 row)
             

   • 创建订阅后，它会将源表中的所有数据加载到目标表。对目标表运行 SELECT 查询以验证初始数据是否已加载：

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       1000
     (1 row)
             

6. 验证源数据库中的复制槽

   在目标数据库中创建订阅会在源数据库中创建复制槽。通过对源数据库运行以下 SELECT 查询来验证复制槽详细信息：

   
   source=> SELECT * FROM pg_replication_slots;
    
   slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size
   ----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------
   testsub   | pgoutput | logical   | 115048 | source   | f         | t      |        846 |      |         6945 | 58/B4000568 | 58/B40005A0         | reserved   |
   (1 row)
           

7. 测试复制

   • 通过在源表中插入行，测试源表中的数据更改是否复制到目标表：

     
     source=> INSERT INTO testtab VALUES(generate_series(1001,2000));
     INSERT 0 1000
     
     source=> SELECT count(*) FROM testtab; 
      count
     -------
       2000
     (1 row)
           

   • 验证目标表中的行数，以确认正在复制新的插入内容：

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       2000
     (1 row)
           

8. 添加表后刷新订阅

   • 在向现有发布添加新表时，务必刷新订阅才能使更改生效：

     
     ALTER SUBSCRIPTION <subscription_name> REFRESH PUBLICATION;
           

   • 此命令从发布者那里提取缺失的表信息，并开始复制自创建或上次刷新订阅以来添加到已订阅发布中的表。