Logical Replication（逻辑复制）

文档用途

了解逻辑复制

详细信息

1、逻辑复制
逻辑复制一种根据数据对象的复制标识 (通常是主键) 复制数据对象及其更改的方法。不同于物理复制，逻辑复制允许对数据复制和安全性进行细粒度控制，如实现部分表复制。Postgresql同时支持物理复制和逻辑复制。
逻辑复制使用发布和订阅模型, 其中一个或多个订阅者订阅发布者节点上的一个或多个发布。订阅者从他们订阅的发布中提取数据, 然后可能重新发布数据, 以允许级联复制或更复杂的配置。
2、适用场景

1)对单个数据库或部分表进行增量同步复制
2)多库数据采集、集中
3)跨大版本同步，跨版本升级
4)实现跨平台同步
5)不支持ddl，只支持基本表
逻辑复制是一种基于复制标识 (通常是主键) 复制数据对象及其更改的方法。逻辑复制由订阅（subscription）和发布（publication）`组成。

3、发布（publication）和发布者（publisher）介绍
发布可以选择INSERT, UPDATE, and DELETE任意组合，默认是所有操作类型
发布的表必须配置‘replica identity’通常是主键！也可以是唯一索引，如果没有合适的key，复制标识可以设置为‘full’这意味着整个行是key，但这样效率非常低。
发布者配置了‘replica identity’,订阅者也需要配置相同的复制标识。如果没有‘replica identity’update，delete会收到影响，但insert不会
一个发布（publication）可对应多个订阅（subscribers）
发布使用以下命令创建

CREATEPUBLICATION

表可以使用以下语句动态的添加和删除从发布中

ALTERPUBLICATION.ADDTABLE/DROPTABLE

4、订阅（Subscription）和订阅者（subscriber）介绍

订阅是复制流的下游端，订阅者也可以作为发布者，级联发布。行为和发布者一样
一个订阅者可以有多个订阅，但是要确保订阅对象不重叠
每个订阅都需要一个replication slot接收change，初始化数据时需要额外的临时复制槽
逻辑复制的订阅者可以是同步流复制的standby端
关于订阅有相关的数据库字典视图catalog.pg_subscription
订阅使用以下的语句创建

CREATESUBSCRIPTION

订阅停止使用/重用可使用以下的语句

ALTERSUBSCRIPTION

移除使用以下的语句

DROPSUBSCRIPTION.

订阅被删除重建后，同步会中断，数据需要重新初始化同步

注意：

模式对象的定义（用户）无法被复制，发布的表结构必须在订阅端已存在，只能用于复制常规表，如无法复制视图view
发布和订阅的表名必须相同
列的名字必须相同，列顺序可以不同，但列类型必须相同，目标端可以有额外的列，它们将被默认值填充

5、违反约束Conflicts

逻辑复制就是普通的dml操作，违反约束逻辑复制将会停止，这叫做conflict。
当复制update or delete操作时，丢失匹配的数据时不会产生冲突，操作将会被跳过！
可以通过更改订阅服务器上的数据来完成该解决方案, 使其不会与传入的更改冲突, 也可跳过与现有数据冲突的事务。通过调用 pg_replication_origin_advance () 函数与订阅名称对应的 node_name 和位置, 可以跳过该事务。
源的当前位置可以在 pg_replication_origin_status 系统视图中看到。

6、限制Restrictions

1 The database schema and DDL commands are not replicated.
2 Sequence data is not replicated
3 TRUNCATE commands are not replicated.
4 large objects are not replicated
5 Replication is only possible from base tables to base tables.

7、架构Architecture

先初始化快照数据，然后同步数据库操作。
逻辑复制架构类似与物理流复制。通过“walsender” and “apply” processes实现，walsender process读取wal 逻辑解码，此处会加载标准逻辑解码插件pgoutput、pgoutput 转换和过滤wal数据，依据发布（publication）定义,数据通过流复制协议传输到备端apply process接收并应用更改到对应的表

8、监控Monitoring

The monitoring information about subscription is visible in pg_stat_subscription

9、Security（安全和权限）

The role used for the replication connection must have the REPLICATION attribute (or be a superuser). Access for the role must be configured in pg_hba.conf and it must have the LOGIN attribute.

10、Quick Setup

1 First set the configuration options in postgresql.conf:
wal_level = logical
The other required settings have default values that are sufficient for a basic setup.
2 pg_hba.conf needs to be adjusted to allow replication (the values here depend on your actual network configuration and user you want to use for connecting):
host all repuser 0.0.0.0/0 md5
3 Then on the publisher database:

CREATEPUBLICATION mypubFORTABLEusers,departments;

4 And on the subscriber database:

CREATESUBSCRIPTION mysub CONNECTION'dbname=foo host=bar user=repuser'PUBLICATION mypub;

The above will start the replication process, which synchronizes the initial table contents of the tables users and departments and then starts replicating incremental changes to those tables.

11、事例

publisher

发布节点配置

1 修改参数

altersystemsetwal_level=logical;altersystemsetmax_replication_slots=8;altersystemsetmax_wal_senders=10;--larger than set max_replication_slotspg_ctl restart

2 手动创建逻辑复制槽（无需创建，这里仅描述复制操的作用）

selectpg_create_logical_replication_slot('logical_slot1','test_decoding');selectslot_name,plugin,slot_type,database,active,restart_lsnfrompg_replication_slots;SELECTpg_drop_replication_slot('test_slot');

注：test_decode是pg内置的一个逻辑decode插件
物理复制槽的主要作用是避免主库可能覆盖并循环使用还没有发给备库的wal日志。逻辑复制槽主要用来存放解析的wal

查看逻辑复制槽解析的数据变化

select*frompg_logical_slot_get_changes('logical_slot1',null,null);lsn|xid|data-----+-----+------(0rows)

注意：此函数捕获数据后，数据将被消耗掉，因此此函数查询仅能显示一次。pg_logical_slot_peek_changes函数可获取逻辑复制槽所有解析的数据，但只能显示pg_logical_slot_get_changes没有消耗的数据。也可使用pg_recvlogical捕获数据变化
$ pg_recvlogical -d postgres --slot logical_slot1 --start -f -v
3 发布节点创建逻辑复制用户

createuserlogicalrepreplicationlogin encrypted password'logicalrep';

4 发布节点创建发布

createpublication pub1fortablet_lr1;orcreatepublication pub1;

for table 关键字制定加入到发布的表列表，仅支持普通表，临时表，外部表，视图，物化视图分区表暂不支持，如果想将分区表添加到发布中，需逐个添加分区表分区到发布
for all tables，发布包括全库，包括新建表

查询发布

select*frompg_publication;pubname|pubowner|puballtables|pubinsert|pubupdate|pubdelete---------+----------+--------------+-----------+-----------+-----------pub1|10|f|t|t|t(1row)

pubowner，发布的属主，与pg_user视图中的usersysid关联
5 配置replication网络访问控制

pg_hba.conf hostallrepuser0.0.0.0/0md5orhostreplicationrepuser x.x.6.13/32md5 pg_ctl reload

subscriber
订阅节点配置

1 参数配置

altersystemsetmax_replication_slots=8;altersystemsetmax_logical_replication_workers=8;--taken from max_worker_processes

2 订阅节点创建订阅

createsubscription sub1 connection'host=x.x.6.12 port=5432 dbname=postgres user=logicalrep password=logicalrep'publication pub1;ERROR: couldnotconnecttothe publisher: couldnotconnecttoserver:NoroutetohostIsthe server runningonhost"x.x.6.12"andaccepting TCP/IP connectionsonport5432?

注意：需要先配置发布节点的网络访问控制pg_hba.conf

注意：查询发布节点，订阅在发布节点上创建了一个逻辑复制槽
postgres=# select slot_name,plugin,slot_type,database,active,restart_lsn from pg_replication_slots;
3 订阅节点查询订阅信息

select*frompg_subscription;

添加同步表
此时逻辑复制已经开始，但无同步数据，因为发布中没有对象

publisher
2019-03-15 11:21:12.641 CST,“logicalrep”,“postgres”,1552,“x.x.6.13:47548”,5c8b1a28.610,2,“idle”,2019-03-15 11:21:12 CST,5/0,0,LOG,00000,“logical decoding found consistent point at 0/7E004BB0”,“There are no running transactions.”,“sub1”
2019-03-15 11:25:35.111 CST,1454,5c8b160a.5ae,3,2019-03-15 11:03:38 CST,0,LOG,00000,“received SIGHUP, reloading configuration files”,“”
subscriber
2019-03-15 11:39:53.187 CST,1347,5c8b1e89.543,1,2019-03-15 11:39:53 CST,3/2,0,LOG,00000,“logical replication apply worker for subscription ““sub1"” has started”,”"

1 发布节点创建测试表

\c postgres highgocreatetabletest_lr1(idint,namechar(10));insertintotest_lr1values(1,'aaa');insertintotest_lr1values(2,'bbb');

2 订阅节点创建表结构

\c postgres highgocreatetabletest_lr1(idint,namechar(10));postgres=# select * from test_lr1;id|name----+------(0rows)

3 将表添加到发布

\c postgres pgalterpublication pub1addtablehighgo.test_lr1;

刷新，立即同步

altersubscription sub1 refresh publication;

此时数据没有同步过来，主要是复制用户logicalrep对表所属的schema及表没有读写权限

数据库报错

2019-03-15 11:57:07.734 CST,"logicalrep","postgres",1636,"x.x.6.13:47572",5c8b2293.664,2,"COPY",2019-03-15 11:57:07 CST,3/66,0,ERROR,42501,"permission denied for schema highgo",,,,,,"COPY highgo.test_lr1 TO STDOUT",,,"sub1_16391_sync_16397" 2019-03-15 11:57:12.768 CST,"logicalrep","postgres",1637,"x.x.6.13:47574",5c8b2298.665,1,"idle in transaction",2019-03-15 11:57:12 CST,3/68,0,LOG,00000,"logical decoding found consistent point at 0/7E02C470","There are no running transactions.",,,,,,,,"sub1_16391_sync_16397" 2019-03-15 11:56:59.321 CST,,,1421,,5c8b228b.58d,2,,2019-03-15 11:56:59 CST,5/63,0,ERROR,XX000,"could not start initial contents copy for table ""highgo.test_lr1"": ERROR: permission denied for schema highgo",,,,,,,,,"" 2019-03-15 11:56:59.322 CST,,,1338,,5c8b1e89.53a,11,,2019-03-15 11:39:53 CST,,0,LOG,00000,"worker process: logical replication worker for subscription 16391 sync 16397 (PID 1421) exited with exit code 1",,,,,,,,,""

publisher
授权

grantusageonschemahighgotologicalrep;grantselectonhighgo.test_lr1tologicalrep;

subscriber再次查询数据同步

postgres=# select * from highgo.test_lr1 ;id|name----+------------1|aaa2|bbb(2rows)

logical replication 同步状态
1 主备库有相关的发布和订阅进程

发布
pg 1463 1454 0 11:03 ? 00:00:00 postgres: bgworker: logical replication launcher
pg 1587 1454 0 11:39 ? 00:00:00 postgres: wal sender process logicalrep x.x.6.13(47550) idle
订阅
postgres 1346 1338 0 11:39 ? 00:00:00 postgres: bgworker: logical replication launcher
postgres 1347 1338 0 11:39 ? 00:00:00 postgres: bgworker: logical replication worker for subscription 16391

2 查询发布中的表列表
\dRp+
or
select * from pg_publication_tables;

DML验证
没有REPLICA IDENTITY，不允许更新update、删除delete操作

订阅和发布，insert不受影响
insert

insertintohighgo.test_lr1values(3,'ccc');

update

updatehighgo.test_lr1setname='jason'whereid=1;

ERROR: cannot update table “test_lr1” because it does not have a replica identity and publishes updates
HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.
更改表结构，创建主键，(publisher and subscriber)

altertablehighgo.test_lr1addprimarykey(id);postgres=# update highgo.test_lr1 set name='jason' where id=1;UPDATE1

发布中添加表、删除表

alterpublication pub1addtablehighgo.test_lr1;alterpublication pub1droptablehighgo.test_lr1;

逻辑复制启动和停止

altersubscription sub1disable;altersubscription sub1enable;

立即同步

altersubscription sub1 refresh publication;

重新同步表

重建订阅,重建订阅后，重新同步对应发布中的所有表

altersubscription sub1disable;postgres=# drop subscription sub1;NOTICE: droppedreplicationslot"sub1"onpublisherDROPSUBSCRIPTIONcreatesubscription sub1 connection'host=x.x.6.12 port=5432 dbname=postgres user=logicalrep password=logicalrep'publication pub1;postgres=# select * from highgo.test_lr1;id|name----+------------1|aaa3|ccc2|bbb4|ddd(5rows)