聊城网站建设广州疫情最新新增
PostgreSQL 变化数据捕捉（CDC）
基于CDC（变更数据捕捉）的增量数据集成总体步骤：
1.捕获源数据库中的更改数据
2.将变更的数据转换为您的消费者可以接受的格式
3.将数据发布到消费者或目标数据库
PostgreSQL支持触发器（trigger）和预写日志（WAL）两种CDC机制。
如果您希望 Postgres 数据更改发生时逐行流式传输，则需要逻辑解码或 Postgres 逻辑复制功能。
使用 Postgres 逻辑解码
逻辑解码是 PostgreSQL 的基于日志的 CDC（逻辑复制）的正式名称。逻辑解码使用 PostgreSQL 预写日志的内容来存储数据库中发生的所有活动。
step1：修改PostgreSQL数据库配置postgresql.conf文件
wal_level = logical
max_replication_slots = 10
max_wal_senders = 20
wal_level ：设置为 logical，允许 WAL 日志记录逻辑解码所需的信息。
max_replication_slots ：确保 max_replication_slots >= 使用 WAL 的 PostgreSQL 连接器的数量加上您的数据库使用的其他复制槽的数量。
max_wal_senders ：指定 WAL 的最大并发连接数的参数，确保 max_wal_senders 至少是逻辑复制槽数的两倍。例如，如果您的数据库总共使用 10 个复制槽，则该 max_wal_senders 值必须为 20 或更大。
配置修改以后，需要重复 PostgreSQL 服务生效配置。
step2：为需要同步的数据库（db）创建逻辑复制插槽（replication slot）
关于复制 SQL 函数更多详情，可以参考官方文档（9.26.系统管理函数）介绍： http://www.postgres.cn/docs/12/functions-admin.html
函数：pg_create_logical_replication_slot(slot_name name, plugin name)
返回类型：(slot_name name, lsn pg_lsn)
说明：使用输出插件plugin创建一个名为 slot_name的新逻辑（解码）复制槽。
创建时需要指定逻辑复制插槽名称和输出插件：
SELECT pg_create_logical_replication_slot('replication_slot01', 'test_decoding'); -- 使用 test_decoding 输出插件
SELECT pg_create_logical_replication_slot('replication_slot01', 'pgoutput'); -- 使用 pgoutput 输出插件
注意权限：
如果用户没有权限可能会报错：ERROR: must be superuser or replication role to use replication slots
授权：
\c - postgres
ALTER ROLE test REPLICATION; --流复制权限
\du
验证插槽是否创建成功：
SELECT * FROM pg_replication_slots;
testdb01=> SELECT * FROM pg_replication_slots;
slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn
--------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
replication_slot01 | test_decoding | logical | 16437 | testdb01 | f | f | | | 585 | 0/1807EA8 | 0/1807EE0
replication_slot02 | pgoutput | logical | 16437 | testdb01 | f | f | | | 585 | 0/1809CD8 | 0/1809D10
(2 rows)
testdb01=>
注意：
1）逻辑复制插槽名称：每个逻辑复制插槽都有一个名称，创建时需要指定名称，可以包含小写字母、数字和下划线
2）输出插件：创建逻辑复制插槽时需要指定输出插件，输出插件有：test_decoding、pgoutput、wal2json等
test_decoding 输出插件 ：PostgreSQL 9.4+原生附带了 test_decoding 输出插件（$PG_HOME/lib 目录下对应有 test_decoding.so），如果您的消费者支持 test_decoding ，则可以使用 test_decoding 输出插件。
pgoutput 输出插件 ：PostgreSQL 10+原生附带了 pgoutput 输出插件（$PG_HOME/lib 目录下对应有 pgoutput.so），如果您的消费者支持 pgoutput ，则可以使用 pgoutput 输出插件。pgoutput插件输出的是二进制的数据
wal2json 输出插件 ：wal2json 是另一个流行的逻辑解码输出插件，PostgreSQL原生不携带、需要数据库服务单独安装插件才能使用。wal2json输出的是json格式的数据
step3：为数据库中的所有表或指定表创建一个发布。如果以指定表方式创建发布则后面可以在发布中管理（添加、删除）表
帮助说明：\h create publication
DROP PUBLICATION IF EXISTS pub01;
CREATE PUBLICATION pub01 FOR TABLE test01;
或者
DROP PUBLICATION IF EXISTS pub01;
CREATE PUBLICATION pub01 FOR TABLE test01, test02, test03;
或者
DROP PUBLICATION IF EXISTS pub01;
CREATE PUBLICATION pub01 FOR ALL TABLES;
查看创建发布的结果：
testdb01=> select * from pg_publication;
oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
-------+---------+----------+--------------+-----------+-----------+-----------+-------------
24629 | pub01 | 16436 | f | t | t | t | t
(1 row)
也可以执行查看：
testdb01=> \dRp
List of publications
Name | Owner | All tables | Inserts | Updates | Deletes | Truncates
-------+-------+------------+---------+---------+---------+-----------
pub01 | test | f | t | t | t | t
(1 row)
创建发布时，还可以选择在发布中包含哪些操作。例如，下面仅创建table01表 & 仅包含INSERT和UPDATE的发布：
CREATE PUBLICATION pub01 FOR TABLE table01 WITH (publish = 'INSERT, UPDATE');
注意：
非超级用户创建发布只能指定表创建、不能 FOR ALL TABLES 创建，否则报错：ERROR: must be superuser to create FOR ALL TABLES publication
FOR ALL TABLES 只允许超级用户创建发布，或者创建后由超级用户修改已创建的发布的配置开启foralltables。
testdb01=> \c - postgres
testdb01=# update pg_publication set puballtables=true where pubname is not null; -- 设置puballtables开关
testdb01=# select * from pg_publication;
oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
-------+---------+----------+--------------+-----------+-----------+-----------+-------------
24629 | pub01 | 16436 | t | t | t | t | t
(1 row)
step4：验证指定的表是否在发布中
testdb01=> SELECT * FROM pg_publication_tables WHERE pubname='pub01';
注意：创建发布不会开始复制。它只为未来的订阅者定义一个分组和过滤逻辑。
step5：查看逻辑复制效果：
获取结果的相关命令：
pg_logical_slot_get_changes : 查询并删除数据。仅在第一次返回结果，多次调用可能会返回空结果集，这意味着当get命令执行时，结果会被提供和删除，这增强了我们编写使用这些事件创建表副本的逻辑的能力。
函数：pg_logical_slot_get_changes(slot_name name, upto_lsn pg_lsn, upto_nchanges int, VARIADIC options text[])
返回类型：(lsn pg_lsn, xid xid, data text)
说明：返回槽slot_name中的改变，从上一次已经被消费的点开始返回。 如果upto_lsn和upto_nchanges为 NULL，逻辑解码将一 直继续到 WAL 的末尾。如果upto_lsn为非 NULL，解码将只包括那些在指 定 LSN 之前提交的事务。如果upto_nchanges为非 NULL， 解码将在其产生的行数超过指定值后停止。不过要注意， 被返回的实际行数可能更大，因为对这个限制的检查只会在增加了解码每个新的提交事务产生 的行之后进行。
pg_logical_slot_get_binary_changes：变化数据以bytea返回。
pg_logical_slot_peek_changes : 只查询不删数据。多次调用每次都会返回相同的结果。是另一个 PostgreSQL 命令，用于在不使用 WAL 条目的情况下查看更改。
函数：pg_logical_slot_peek_changes(slot_name name, upto_lsn pg_lsn, upto_nchanges int, VARIADIC options text[])
返回类型：(lsn text, xid xid, data text)
说明：行为就像pg_logical_slot_get_changes()函数， 不过改变不会被消费， 即在未来的调用中还会返回这些改变。
pg_logical_slot_peek_binary_changes：变化数据以bytea返回。
关于复制 SQL 函数更多详情，可以参考官方文档（9.26.系统管理函数）介绍： http://www.postgres.cn/docs/12/functions-admin.html
测试表中I/D/U变更数据操作，然后查看逻辑复制效果：
testdb01=> SELECT * FROM pg_logical_slot_peek_changes('replication_slot01', NULL, NULL); -- 只查询不删数据
......
testdb01=> SELECT * FROM pg_logical_slot_get_changes('replication_slot01', NULL, NULL); -- 查询并删除数据
lsn | xid | data
-----------+-----+------------------------------------------------------------------------------------------
0/180A050 | 586 | BEGIN 586
0/180A050 | 586 | table public.test01: INSERT: id[bigint]:1 info[character varying]:'aa' cnt[integer]:123
0/180A160 | 586 | table public.test01: INSERT: id[bigint]:2 info[character varying]:'bb' cnt[integer]:456
0/180A1E8 | 586 | table public.test01: INSERT: id[bigint]:3 info[character varying]:'cc' cnt[integer]:55
0/180A2A0 | 586 | COMMIT 586
0/180A2A0 | 587 | BEGIN 587
0/180A2A0 | 587 | table public.test01: INSERT: id[bigint]:4 info[character varying]:'uuu' cnt[integer]:66
0/180A358 | 587 | COMMIT 587
0/180A358 | 588 | BEGIN 588
0/180A358 | 588 | table public.test01: DELETE: id[bigint]:2
0/180A3D0 | 588 | COMMIT 588
0/180A3D0 | 589 | BEGIN 589
0/180A3D0 | 589 | table public.test01: UPDATE: id[bigint]:3 info[character varying]:'xxxx' cnt[integer]:55
0/180A458 | 589 | COMMIT 589
(14 rows)
testdb01=> SELECT * FROM pg_logical_slot_get_changes('replication_slot01', NULL, NULL); -- 查询并删除数据
lsn | xid | data
-----+-----+------
(0 rows)
testdb01=>
pgoutput输出插件的结果查看，则执行：
testdb01=> SELECT * FROM pg_logical_slot_peek_binary_changes('replication_slot02', null, null, 'proto_version', '1', 'publication_names', 'pub01'); -- 查询但不删除数据
......
testdb01=> SELECT * FROM pg_logical_slot_get_binary_changes('replication_slot02', null, null, 'proto_version', '1', 'publication_names', 'pub01'); -- 查询并删除数据
lsn | xid | data
-----------+-----+------------------------------------------------------------------------------------------------------------------------------
0/1809DF8 | 585 | \x420000000001809fd0000296c5f556e57d00000249
0/1809DF8 | 585 | \x52000060417075626c696300746573743031006400030169640000000014ffffffff00696e666f00000004130000006800636e740000000017ffffffff
0/1809DF8 | 585 | \x44000060414b00037400000001316e6e
0/1809F40 | 585 | \x44000060414b00037400000001346e6e
0/1809F88 | 585 | \x44000060414b00037400000001336e6e
0/180A000 | 585 | \x43000000000001809fd0000000000180a000000296c5f556e57d
0/180A050 | 586 | \x42000000000180a270000296c5f7bb75300000024a
0/180A050 | 586 | \x49000060414e0003740000000131740000000261617400000003313233
0/180A160 | 586 | \x49000060414e0003740000000132740000000262627400000003343536
0/180A1E8 | 586 | \x49000060414e00037400000001337400000002636374000000023535
0/180A2A0 | 586 | \x4300000000000180a270000000000180a2a0000296c5f7bb7530
0/180A2A0 | 587 | \x42000000000180a328000296c5f7d8abe90000024b
0/180A2A0 | 587 | \x49000060414e0003740000000134740000000375757574000000023636
0/180A358 | 587 | \x4300000000000180a328000000000180a358000296c5f7d8abe9
0/180A358 | 588 | \x42000000000180a3a0000296c5f80ae6ca0000024c
0/180A358 | 588 | \x44000060414b00037400000001326e6e
0/180A3D0 | 588 | \x4300000000000180a3a0000000000180a3d0000296c5f80ae6ca
0/180A3D0 | 589 | \x42000000000180a428000296c5f826073e0000024d
0/180A3D0 | 589 | \x55000060414e000374000000013374000000047878787874000000023535
0/180A458 | 589 | \x4300000000000180a428000000000180a458000296c5f826073e
(20 rows)
testdb01=> SELECT * FROM pg_replication_slots;
slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn
--------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------
replication_slot01 | test_decoding | logical | 16437 | testdb01 | f | f | | | 590 | 0/180A540 | 0/180A578
replication_slot02 | pgoutput | logical | 16437 | testdb01 | f | f | | | 590 | 0/180A540 | 0/180A578
(2 rows)
PG logical replication plug-in ("pgoutput") 输出的消息格式详情可以参考PG官方说明：https://www.postgresql.org/docs/10/protocol-logicalrep-message-formats.html
step6：销毁逻辑复制插槽
SELECT pg_drop_replication_slot('replication_slot01');
使用复制插槽的注意事项：
1）每个插槽只有一个输出插件（创建插槽时由您来选择使用哪个）。
2）每个插槽仅提供来自一个数据库的更改。
3）一个数据库可以有多个插槽。
4）每个数据更改通常在每个插槽中发出一次。
5）但是当 Postgres 实例重新启动时，插槽可能会重新发出更改。消费者必须处理这种情况。
6）未使用的插槽对 Postgres 实例的可用性构成威胁。Postgres 将为这些未使用的更改保存所有 WAL 文件。这可能导致存储溢出。
关于 PostgreSQL WAL 消费者：
可以获取 Postgres 逻辑解码流的任何应用程序都是 PostgreSQL WAL 消费者。
pg_recvlogical：
pg_recvlogical 是一个 PostgreSQL 应用程序，他是 PostgreSQL 原生的逻辑解码工具，它可以管理槽并使用槽中的流。它包含在 Postgres 发行版中，因此它可能已经随 PostgreSQL 一起安装，在 $PG_HOME/bin 目录下。
pg_recvlogical 使用默认的 test_decoding 逻辑解码插件，pg安装完后会在pg安装目录的lib目录下创建test_decoding链接库文件。
pg_recvlogical 使用介绍：
创建逻辑复制插槽：
$ pg_recvlogical --create-slot -S replication_slot01 -d testdb01
启动复制槽解码（启动后会实时的将日志解码到制定的文件中，也可以不启动、等需要解码时再启动解码）
$ pg_recvlogical --start -S replication_slot01 -d testdb01 -f replication_slot01_decoding.log &
查看解码结果：
$ cat replication_slot01_decoding.log
查看数据库lsn：
postgres=# select pg_current_wal_lsn();
使用pg_recvlogical进行日志区间解码：
$ pg_recvlogical --start -S replication_slot01 -d testdb01 -I 4C/180B020 -E 4C/180B1EE -f ret.log