【postgresql初级使用】逻辑复制是对数据库对象进行复制，非常灵活的完成数据归集与分发
逻辑复制介绍
专栏内容：
- postgresql使用入门基础
- 手写数据库toadb
- 并发编程
个人主页：我的主页
管理社区：开源数据库
座右铭：天行健，君子以自强不息；地势坤，君子以厚德载物.
✅ 🔥🔥🔥重大消息🔥🔥🔥 ❤️❤️❤️❤️ 关注公众号【开源无限】可免费领取《手写数据库内核toadb》源代码一份 ❤️❤️❤️❤️
一、概述
在我们使用数据库时，往往需要感知数据库中某些数据库对象的变化，比如表中insert/update操作使数据发生了变化。
为了及时得到数据的变化，我们常常会开启一个循环和定时器，不断的查询比较，这个工作非常耗时和容易出错，还不是很准确，令人非常头疼。
在Postgresql中有两种实时的复制模式：
- 一种对文件内容的复制，也就是文件中二进制数据直接复制，也称为物理复制；
- 另一种是对数据库对象的复制，数据库对象比如table, database都是逻辑概念，所以也称为逻辑复制；
逻辑复制本身就是基于数据库对象的变化，当有变化时就需要产生复制事件，这一特性刚好就可以用来作为数据库对象的变化事件，这样只需要订阅这一事件即可，由数据库来检查数据库对象的变化，并通知我们，即省力就及时。
本节就来聊一下逻辑复制的使用方法。
二、 部署逻辑备库
先来看一下如何使用postgresql的逻辑复制，我们通过部署一个备库的方式来介绍。
在postgresql 中通过发布订阅的方式来使用，基本步骤如下：
- 在发布端定义需要关注的数据库对象；
- 发布端在该数据库对象变化时，就会发布相应变化；
- 而需要感知的客户端，需要订阅对应的通知；
- 当有变化发布时，就会通知到所有订阅者；
2.1 配置集簇
- 初始化集簇
先来准备两个postgresql集簇 pgA, pgB，使用initdb命令进行初始化。
[senllang@hatch postgres]$ /opt/postgres/bin/initdb -D pgA
[senllang@hatch postgres]$ /opt/postgres/bin/initdb -D pgB
pgA作为主集簇，而pgB作为它的逻辑备库。
- 配置参数
配置WAL的级别为逻辑复制，默认是物理复制。
vi pgB/postgresql.conf
wal_level = logical
另外两个参数，由备库的数量来决定，默认值已经满足本次部署，不作修改。
#max_wal_senders = 10
#max_replication_slots = 10
当然，如果部署在不同机器上时，还需要在pg_hba.conf中增加用户的访问权限，请参考权限控制章节。
- 启动集簇
现在将两个集簇都启动，pgA使用默认5432端口启动，因为在同一台机器上，所以pgB启动时用-o指定参数端口号(-p)为5433。
[senllang@hatch postgres]$ /opt/postgres/bin/pg_ctl -D pgA -l logfile start
waiting for server to start.... done
server started
[senllang@hatch postgres]$ /opt/postgres/bin/pg_ctl -D pgB -o "-p 5433" -l logfile start
waiting for server to start.... done
server started
将集簇A定为发布者，集簇B为订阅者，作为A的备库。
2.2 发布
将集簇A定为发布者，集簇B为订阅者，作为A的备库。
我们将在主库和备库上都创建student表，来作为演示。
先登陆主库，创建数据表，并建立发布源。
[senllang@hatch postgres]$ /opt/postgres/bin/psql -d postgres -U postgres
psql (17devel)
Type "help" for help.
postgres=# create table student(sid int, sname varchar);
CREATE TABLE
此处将表student所有操作，都会发布出去，了解发布者的信息可以作如下查询。
postgres=# select a.*, b.tablename,attnames from pg_publication as a left join pg_publication_tables as b on a.pubname = b.pubname;
oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot | tablename | attnames
-------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------+-----------+-------------
16400 | pub_stu | 10 | f | t | t | t | t | f | student | {sid,sname}
(1 row)
pub_stu发布源，对应发布的表为student的所有列，同时对于insert/update/delete/truncate操作都会发布。
此时发布者已经创建好了，静等订阅者了。
2.3 订阅
登陆备库，这里备库的端口号为5433，登陆时需要指定。
在备库上需要创建同样的表student，作为接收数据表。
在备库上创建订阅，当主库有事件发生时，就会接收，并作出同样的变化。
[senllang@hatch postgres]$ /opt/postgres/bin/psql -d postgres -U postgres -p 5433
psql (17devel)
Type "help" for help.
postgres=# create table student(sid int, sname varchar);
CREATE TABLE
创建订阅，与发布源pub_stu进行关联。
postgres=# create subscription sub_stu CONNECTION 'host=127.0.0.1 dbname=postgres user=postgres' publication pub_stu;
NOTICE: created replication slot "sub_stu" on publisher
CREATE SUBSCRIPTION
查询订阅信息。
postgres=# select * from pg_subscription;
oid | subdbid | subskiplsn | subname | subowner | subenabled | subbinary | substream | subtwophasestate | subdisableonerr | subpasswordrequired | subrunasowner | subfailover |
subconninfo | subslotname | subsynccommit | subpublications | suborigin
-------+---------+------------+---------+----------+------------+-----------+-----------+------------------+-----------------+---------------------+---------------+-------------+---
-------------------------------------------+-------------+---------------+-----------------+-----------
16399 | 5 | 0/0 | sub_stu | 10 | t | f | f | d | f | t | f | f | ho
st=127.0.0.1 dbname=postgres user=postgres | sub_stu | off | {pub_stu} | any
(1 row)
这里就可以看到当前集簇中的所有订阅者信息。
查看订阅状态。
postgres=# select * from pg_stat_subscription;
subid | subname | worker_type | pid | leader_pid | relid | received_lsn | last_msg_send_time | last_msg_receipt_time | latest_end_lsn | latest_end_tim
e
-------+---------+-------------+---------+------------+-------+--------------+-------------------------------+-------------------------------+----------------+----------------------
---------
16399 | sub_stu | apply | 2041851 | | | 0/1630BE8 | 2024-11-13 09:03:07.365009+08 | 2024-11-13 09:03:07.365055+08 | 0/1630BE8 | 2024-11-13 09:03:07.3
65009+08
(1 row)
可以看到，订阅已经成功了，有最后发送和接收消息的时间，还有最新的LSN。
2.4 验证
现在来看一下效果，在主库的student进行增删改查的操作，看是否能同步到备库中。
- 新增数据
登陆主库，插入两条数据。
[senllang@hatch postgres]$ /opt/postgres/bin/psql -d postgres -U postgres
psql (17devel)
Type "help" for help.
postgres=# insert into student values(10,'10');
INSERT 0 1
postgres=# insert into student values(11,'11');
INSERT 0 1
在备库进行查看。
[senllang@hatch postgres]$ /opt/postgres/bin/psql -d postgres -U postgres -p 5433
psql (17devel)
Type "help" for help.
postgres=# select * from student ;
sid | sname
-----+-------
10 | 10
11 | 11
(2 rows)
已经同步过来了，时间还是非常快。
- 修改数据
在主库进行修改数据，修改sid=10的sname字段。
postgres=# update student set sname='xiaohua' where sid=10;
ERROR: cannot update table "student" because it does not have a replica identity and publishes updates
HINT: To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.
没有修改成功，根据报错信息，因为有复制的存在，而表里没有定义复制的标识条符，表上也没有主键等唯一索引，就没有办法唯一标识一行数据，主备之间复制时就不能进行映射。
这里解决办法是两种：一种是创建复制标识符，指定值不同的某一列；另一种是在表上定义唯一性带索引。
我们在表上增加主键，就可以继续修改了。
postgres=# alter table student add primary key (sid);
ALTER TABLE
postgres=# update student set sname='xiaohua' where sid=10;
UPDATE 1
可以修改成功了，再来看一下备库是否同步成功。
备库此时还没有进行同步，因为备库还没有主键，搞不清楚sid=10的数据是新增呢，还是修改。
所以先要在备份上添加主键约束。
postgres=# select * from student ;
sid | sname
-----+-------
10 | 10
11 | 11
(2 rows)
postgres=# alter table student add primary key (sid);
ALTER TABLE
再次查询时，发现已经同步过来了。
postgres=# select * from student ;
sid | sname
-----+---------
11 | 11
10 | xiaohua
(2 rows)
其它操作也是类似，不再这里一一演示。
五、总结
在PostgreSQL中逻辑复制是一个非常实用的功能，因为它可以指定数据库对象进行复制，甚至可以指定到列，没有指定的列可以不复制，这就非常灵活。
可以用于数据的汇集，也可以用于数据的分类，特定数据的分发或备份等。
当然，逻辑复制非常灵活，在部署时一定要先设计好数据复制的路径，避免产生复制的循环，比如A复制到B，B又复制到A，这样数据就会越来越多。
结尾
非常感谢大家的支持，在浏览的同时别忘了留下您宝贵的评论，如果觉得值得鼓励，请点赞，收藏，我会更加努力！
作者邮箱：study@senllang.onaliyun.com
如有错误或者疏漏欢迎指出，互相学习。
注：未经同意，不得转载！