MogDB 3.0.0 新特性——发布订阅

MogDB3.0.0新特性——发布订阅

概述

特性简介

发布订阅基于逻辑复制实现，其中有一个或者更多订阅者订阅一个发布者节点上的一个或者更多发布。订阅者从它们所订阅的发布拉取数据。实现集群间的数据实时同步。

客户价值

发布订阅的典型使用场景是：

• 在一个数据库或者一个数据库的子集中发生更改时，把增量的改变发送给订阅者。
• 在更改到达订阅者时引发触发器。
• 把多个数据库联合到单一数据库中（例如用于分析目的）。

特性描述

发布者上的更改会被实时发送给订阅者。订阅者以与发布者相同的顺序应用那些数据，这样在一个订阅中能够保证发布的事务一致性。这种数据复制的方法有时候也被称为事务性复制。

订阅者数据库的行为与任何其他MogDB实例相同，并且可以被用作其他数据库的发布者，只需要定义它自己的发布。当订阅者被应用当作只读时，单一的订阅中不会有冲突。在另一方面，如果应用或者对相同表集合的订阅者执行了其他的写动作，冲突可能会发生。

参考官方文档

流程

1. 发布端配置相关参数，创建发布（wal_level = logical）
2. 订阅端建表，创建订阅（需要指定发布端的IP，PORT，用户名，密码，发布名称）
3. 订阅端发送链接请求，建立链接（建立类似主备的链接）
4. 订阅端向发布端发送命令，创建复制槽（复制槽的作用是记录wal日志解析位置）
5. 创建订阅的事务提交，launcher线程被唤醒，发现有新的订阅产生，启动apply线程
6. apply线程通过copy命令复制基础数据；基础数据复制完成后，开启流复制同步增量数据，同步格式为二进制格式。
7. 对于增量数据：订阅端将二进制格式的数据组装成tuple直接调用存储层heap接口，对表进行IUD操作

原理

发布订阅实际上是对OutputPluginCallbacks 众多回调函数的实现。其默认实现是pgoutput.cpp 。

/*
 * Output plugin callbacks
 */
typedef struct OutputPluginCallbacks {
    LogicalDecodeStartupCB startup_cb;
    LogicalDecodeBeginCB begin_cb;
    LogicalDecodeChangeCB change_cb;
    LogicalDecodeCommitCB commit_cb;
    LogicalDecodeAbortCB abort_cb;
    LogicalDecodePrepareCB prepare_cb;
    LogicalDecodeShutdownCB shutdown_cb;
    LogicalDecodeFilterByOriginCB filter_by_origin_cb;
} OutputPluginCallbacks;

1
2
3
4
5
6
7
8
9
10
11
12
13
14

回调消息处理流程

• ‘B’
  • 读取事务信息，事务信息包括结束LSN位置、事务提交时间、事务id
  • 记录最新LSN位置
  • 标记发送端当前在事务中
  • 报告活跃状态为RUNNING
• ‘C’
  • 读取事务提交信息，包括提交LSN、结束LSN、事务提交时间
  • 更新复制源LSN为事务结束LSN，更新复制源时间戳为提交时间
  • CommitTransactionCommand
• ‘I’
  • 解析消息体，读取rel_oid，newtup信息
  • open relation并切初始化executor state、tuple slot
  • 执行INSERT（tableam_tuple_insert)
• ‘U’
  • 解析消息体，读取rel_oid，oldtup，newtup信息
  • open relation并初始化executor state、tuple slot
  • 是否能在relation中找到oldtup?如果不能找到，无需进一步处理
  • 执行UPDATE(tableam_tuple_update)
• ‘D’

  • 解析消息体，读取rel_oid，newtup信息

  • open relation并初始化executor state、tuple slot

  • 是否能在relation中找到oldtup?如果不能找到，无需进一步处理

  • 执行DELETE(tableam_tuple_delete)

演示

环境

frank@LAPTOP-4OF1323N:/$ docker run --name mogdb --privileged=true -d -e GS_PASSWORD=Enmo@123  -v /mogdb:/var/lib/mogdb  -p 15432:5432 -p 15433:5433 swr.cn-north-4.myhuaweicloud.com/mogdb/mogdb:3.0.0
a925573d1d38a472b7749e22aa3bc99974a82cae6f91914fe37067701871da65
frank@LAPTOP-4OF1323N:/$ docker run --name mogdb3 --privileged=true -d -e GS_PASSWORD=Enmo@123  -v /mogdb3:/var/lib/mogdb  -p 35432:5432 -p 35433:5433 swr.cn-north-4.myhuaweicloud.com/mogdb/mogdb:3.0.0
0e75166b099ac2fed0a466c7e26f2b1620b02831c09389d5348768ff1184b74d

1
2
3
4
5

pub端

设置参数postgres.conf

wal_level = logical
max_replication_slots = 8 ##大于订阅数+物理复制槽数量+少量剩余用作临时快照复制
max_wal_senders = 10 #大于max_replication_slots+物理复制的备机数量
1
2
3

设置白名单pg_hba.conf

# TYPE  DATABASE        USER            ADDRESS                 METHOD

# "local" is for Unix domain socket connections only
local   all             all                                     trust
# IPv4 local connections:
host    all             all             127.0.0.1/32            trust
# IPv6 local connections:
host    all             all             ::1/128                 trust
# Allow replication connections from localhost, by a user with the
# replication privilege.
#local   replication     omm                                trust
#host    replication     omm        127.0.0.1/32            trust
#host    replication     omm        ::1/128                 trust

host all all 0.0.0.0/0 md5
host replication all 0.0.0.0/0 md5
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

创建角色/用户，这里要有SYSADMIN/OPRADMIN，REPLICATION权限，

MogDB=#CREATE ROLE pub_sub_user WITH SYSADMIN REPLICATION LOGIN PASSWORD 'pub_sub@123';
NOTICE:  The encrypted password contains MD5 ciphertext, which is not secure.
CREATE ROLE
MogDB=#
1
2
3
4

创建要发布的测试表

MogDB=#create table pub_sub(i serial, name varchar);
NOTICE:  CREATE TABLE will create implicit sequence "pub_sub_i_seq" for serial column "pub_sub.i"
CREATE TABLE
1
2
3

赋权

MogDB=#CREATE PUBLICATION user_pub FOR TABLE pub_sub;
CREATE PUBLICATION
MogDB=#GRANT SELECT ON TABLE pub_sub TO pub_sub_user;
GRANT
1
2
3
4

查看发布设置，检验发布是否成功

sub端

创建于pub端相同结构的测试表

MogDB=#create table pub_sub(i serial, name varchar);
NOTICE:  CREATE TABLE will create implicit sequence "pub_sub_i_seq" for serial column "pub_sub.i"
CREATE TABLE

1
2
3
4

创建订阅

MogDB=#CREATE SUBSCRIPTION user_sub CONNECTION 'dbname=postgres host=172.17.0.1 port=15433 user=pub_sub_user password=pub_sub@123' PUBLICATION user_pub;
NOTICE:  created replication slot "user_sub" on publisher
CREATE SUBSCRIPTION

1
2
3
4

查看订阅属性

两端数据同步

Q&A

Q1：

MogDB=#CREATE SUBSCRIPTION user_sub CONNECTION 'dbname=postgres host=172.17.0.1 port=15433 user=pub_sub_user password=pub_sub@123' PUBLICATION user_pub;
ERROR:  No key file subscription.key.cipher
HINT:  Please create subscription.key.cipher file with gs_guc and gs_ssh, such as :gs_ssh -c "gs_guc generate -S XXX -D $GAUSSHOME/bin -o subscription"

1
2
3
4

A1：sub端

omm@4c1187b588b6:~$ gs_guc generate -S pub_sub@123 -D $GAUSSHOME/bin -o subscription
The gs_guc run with the following arguments: [gs_guc -S ******** -D /usr/local/mogdb/bin -o subscription generate ].
gs_guc generate -S ***
omm@4c1187b588b6:~$

1
2
3
4
5

Q2：

MogDB=#CREATE SUBSCRIPTION user_login_sub CONNECTION 'dbname=postgres host=172.17.0.1 port=15432 user=pub_sub_user password=pub_sub@123' PUBLICATION user_pub;
WARNING:  apply worker could not connect to the remote server
ERROR:  could not connect to the publisher

1
2
3
4

A2:

1 端口未映射到宿主机（docker部署需要注意）。

2 查看信息错误信息

select lwtid from dbe_perf.thread_wait_status w , pg_stat_activity b
where w.sessionid = b.sessionid and query_start > sysdate - 1/86000 AND b.application_name = 'gsql';
1
2

strace -s 8192 -o rep.trc -p <lwtid>
1

用于连接发布端且具有系统管理员权限（SYSADMIN）或者运维管理员权限（OPRADMIN）的用户名和密码。password需要加密，创建订阅前需要在订阅端执行gs_guc generate -S xxxxxx -D $GAUSSHOME/bin -o subscription。