=pg_recvlogical使用===

pg_recvlogical -d postgres -U aaron --slot=test_database --start -f recv.sql

pg_logical_slot_get_changes()函数展示解析数据并消费掉;

pg_logical_slot_peek_changes()函数展示解析数据不会消费掉；

pg_recvlogical，pg自带的工具，能消费复制槽内的数据，相当于逻辑复制的下游。对应物理wal接收工具为 pg_receivewal

=创建复制槽===
SELECT pg_create_logical_replication_slot(‘logical_test’, ‘test_decoding’);
select * from pg_logical_slot_peek_changes(‘logical_test’,null,null);
select * from pg_logical_slot_get_changes(‘logical_test’,null,null);

select pg_replication_slot_advance();推进复制槽

SELECT pg_drop_replication_slot(‘slot_name’);

发布订阅==
订阅发布默认走的pgoutput解析插件。

CREATE PUBLICATION test_all_tables FOR ALL TABLES;
DROP PUBLICATION test_all_tables;
create table test(id int,name varchar(20));
insert into test values(generate_series(100000,500000),‘wjf’);

//删除订阅
alter subscription test disable;
alter subscription test set (slot_name =none);
DROP SUBSCRIPTION test;

====启动订阅服务，另一个pg
DROP SUBSCRIPTION test;

SELECT * FROM pg_catalog.pg_publication;

DROP SUBSCRIPTION sbtest;
SELECT * FROM pg_subscription;

CREATE SUBSCRIPTION test CONNECTION ‘host=localhost port=5432 dbname=postgres user=aaron password=admin’ PUBLICATION test_all_tables;

CREATE SUBSCRIPTION sbtest CONNECTION ‘host=localhost port=5432 dbname=postgres user=aaron password=admin’
PUBLICATION test_all_tables with (create_slot = false,slot_name = test);