PostgreSQL 逻辑复制模块（二）

--创建复制槽
test=# SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
  slot_name    |   lsn    
-----------------+-----------
regression_slot | 0/16569D8
(1 row)


-- 查看复制槽
test=# select * from pg_replication_slots ;


--插入一条数据到test01表中
test=# insert into test01 values(1,'test_decoding use by SQL');
INSERT 0 1


--查看解析的sql
test=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
  lsn     | xid  |                                             data                                              
------------+------+-----------------------------------------------------------------------------------------------
0/340001F8 | 1673 | BEGIN 1673
0/340001F8 | 1673 | table public.test01: INSERT: id[integer]:1 name[character varying]:'test_decoding use by SQL'
0/34000628 | 1673 | COMMIT 1673

# 在test库中创建默认test_decoding类型的复制槽 test2
[postgres@bogon ~]$ pg_recvlogical -d test --slot=test2 --create-slot


# 输出逻辑解码的信息到当前目录下的 ld.out 文件中
[postgres@bogon ~]$ pg_recvlogical -d test --slot=test2 --start -f ld.out &
[1] 19807


#从test库中插入一条数据
[postgres@bogon ~]$ psql -d test -c "insert into test01 values(2,'test_decoding use by pg_recvlogical');" 
INSERT 0 1


#查看ld.out 中的信息
[postgres@bogon ~]$ cat ld.out 
BEGIN 1674
table public.test01: INSERT: id[integer]:2 name[character varying]:'test_decoding use by pg_recvlogical'
COMMIT 1674

# 在test库中创建wal2json类型的复制槽 test_slot_wal2json
[postgres@bogon ~]$ pg_recvlogical -d test --slot test_slot_wal2json --create-slot -P wal2json


# 输出逻辑解码的信息到当前目录下的 ld_wal2json.out 文件中
[postgres@bogon ~]$ pg_recvlogical -d test --slot test_slot_wal2json --start -o pretty-print=1 -f ld_wal2json.out &


#从test库中插入一条数据
[postgres@bogon ~]$ psql -d test -c "insert into test01 values(2,'wal2json use by pg_recvlogical');" 
INSERT 0 1

{
"change": [
{
"kind": "insert",
"schema": "public",
"table": "test01",
"columnnames": ["id", "name"],
"columntypes": ["integer", "character varying"],
"columnvalues": [2, "wal2json use by pg_recvlogical"]
}
]
}

5 Logical Replication 逻辑订阅处理流程解析

5.1 表同步阶段处理流程概述

       订阅端执行CREATE SUBSCRIPTION后，在后台进行表数据同步。每个表的数据同步状态记录在pg_subscription_rel.srsubstate中,一共有4种状态码。

• 'i'：初始化（SUBREL_STATE_INIT）

• 'd'：正在copy数据（SUBREL_STATE_DATASYNC）

• 's'：已同步（SUBREL_STATE_SYNCDONE）

• 'r'：准备好 （普通复制）（SUBREL_STATE_READY）

subdb=# select * from pg_subscription_rel;
srsubid | srrelid | srsubstate | srsublsn  
---------+---------+------------+------------
  29912 |   29904 | r          | 0/3502AA90
(1 row)

       从执行CREATE SUBSCRIPTION开始订阅端的相关处理流程概述如下：

1. 设置每个表的为srsubstate中 'i'（SUBREL_STATE_INIT）；

2. logical replication launcher 进程启动一个 logical replication apply worker 进程；

3. logical replication apply worker 进程连接到订阅端开始接受订阅消息，此时表尚未完成初始同步（状态为 i 或 d ），跳过所有 insert、update 和 delete 消息的处理；

4. logical replication apply worker 进程为每个未同步的表启动 logical replication sync worker 进程（每个订阅最多同时启动max_sync_workers_per_subscription个 sync worker）；

5. logical replication sync worker 进程连接到订阅端并同步初始数据；

• 创建临时复制槽，并记录快照位置

• 设置表同步状态为 'd'（SUBREL_STATE_DATASYNC）

• copy 表数据

• 设置表同步状态为SUBREL_STATE_SYNCWAIT（内部状态），并等待 apply worker 更新状态为SUBREL_STATE_CATCHUP（内部状态）

6. logical replication apply worker 进程更新表同步状态为SUBREL_STATE_CATCHUP（内部状态），记录最新lsn，并等待 sync worker 更新状态为SUBREL_STATE_SYNCDONE；

7. logical replication sync worker 进程完成初始数据同步；

• 检查 apply worker 当前处理的订阅消息位置是否已经走到了快照位置前面，如果是从订阅端接受消息并处理直到追上 apply worker 

• 设置表同步状态为 's'（SUBREL_STATE_SYNCDONE）

• 进程退出

8. logical replication apply worker 进程继续接受订阅消息并处理。

• 接受到 insert、update 和 delete 消息，如果是同步点（进入's' 或 'r' 状态时的 lsn 位置）之后的消息进行应用

• 接受到 commit 消息

• 暂时没有新的消息处理

    1.向发布端发送订阅位置反馈

   2.如果不在事务块里，同步表状态。将所有处于 's'（SUBREL_STATE_SYNCDONE）同步状态的表更新为 'r'（SUBREL_STATE_READY）

5.2 表同步后的持续逻辑复制

       订阅表进入同步状态（状态码是 ‘s’ 或 'r' ）后，发布端的变更都会通过消息通知订阅端；订阅端 apply worker 按照订阅消息的接受顺序（即发布端事务提交顺序）对每个表 apply 变更，并反馈 apply 位置，用于监视复制延迟。

       插入订阅表

insert into test01 values(100,'insert 1 条');

       发布端修改订阅表时，在事务提交时，发布端依次发送下面的消息到订阅端

• B（BEGIN）

• R（RELATION）

• I（INSERT）

• C（COMMIT）

  更新复制源状态表pg_replication_origin_status中的remote_lsn和local_lsn，该位点对应于每个订阅表最后一次事务提交的位置。

• k（KEEPALIVE）

• k（KEEPALIVE）

  2个 keepalive 消息，会更新统计表中的位置：

• 发布端pg_stat_replication:write_lsn,flush_lsn,replay_lsn

pubdb=# select usesysid,usename,application_name,sent_lsn,write_lsn,flush_lsn,replay_lsn,sync_state from pg_stat_replication where usename = 'rep';

• 发布端pg_get_replication_slots():confirmed_flush_lsn

pubdb=# select slot_name,plugin,active_pid,restart_lsn,confirmed_flush_lsn from pg_get_replication_slots() where slot_name = 'sub1';

• 订阅端更新pg_stat_subscription:latest_end_lsn

subdb=# select subid,subname,received_lsn,latest_end_lsn from pg_stat_subscription;

5.3 异常处理

5.3.1 sync worker 进程

1. SQL错误（如主键冲突）：worker 进程异常退出，之后 apply worker 进程创建一个新的 sync worker 重试。错误解除前每5秒重试一次

2. 表被锁：等待

3. 更新或删除的记录不存在：正常执行，检测不到错误，也么没有日志输出（输出一条DEBUG1级别的日志）

5.3.2 apply worker 进程

1. SQL错误（如主键冲突）：apply worker 进程异常退出，之后 logical replication launcher 进程创建一个新的apply worker 重试。错误解除前每5秒重试一次

2. 表被锁：等待

3. 更新或删除的记录不存在：正常执行，检测不到错误，也么没有日志输出（输出一条DEBUG1级别的日志）。

6 逻辑复制适用场景

• 可从多个上游服务器,做数据的聚集和合并

         发布者跟订阅者的关系：一个发布者可以被多个订阅者订阅。多个发布者可以被同一个订阅者订阅。

• PostgreSQL大版本升级，数据直接同步到高版本

         pglogical 对 PostgreSQL 版本升级是一个很实用的工具。能实现以几乎为零的停机时间迁移和升级PostgreSQL。局限性在于pglogical支持的 PostgreSQL 版本。

• 将多个数据库实例的数据，同步到一个目标数据库。例如多个数据库同步到一个大的数据仓库

• 满足业务上需求，实现某些指定表的数据同步

7 逻辑复制注意事项

• 发布节点的WAL_LEVEL参数需要设置成LOGICAL

• 发布节点上逻辑复制用户至少需要REPLICATION角色权限

• 支持一次发布一个数据库中的所有表

• 发布节点上需要发布的表如果需要将UPDATE/DELETE操作同步到订阅节点，需要给发布配置复制标识（复制标识默认为主键，如果没有主键也可以是唯一索引）

• 发布表上的DDL操作不会自动同步到订阅节点，如果发布节点上发布的表执行了DDL操作，需手工给订阅节点的相应表执行DDL，之前如果有没有同步的数据会自动同步

• 当发布者添加新表时，订阅者不能自动的获知，需要将表的SELECT权限赋给逻辑复制用户 （发布端执行 ）

• 订阅者列多于发布者（订阅者的列包含发布者的列）时，发布者插入的数据会同步到订阅者，多的字段内容为 NULL

• 建立逻辑复制后，由于冲突或者表结构变更，导致逻辑复制关系挂起后，通过解决冲突和问题后，逻辑复制关系会恢复，并会同步此时发布者的数据

• 发布者表结构变更后，插入数据（1、字段为之前数据类型，2、字段为变更后的数据类型），逻辑复制挂起，均不能进行数据同步，当解决冲突后，逻辑复制恢复，同步数据