postgresql从库搭建--逻辑复制

postgres=# CREATE USER logical_repl REPLICATION LOGIN CONNECTION LIMIT 8 ENCRYPTED PASSWORD ‘logical_repl‘; 

CREATE ROLE

/** 创建用于逻辑复制的主库 */ 
[postgres@PG32 data]$ createdb sourcedb 
[postgres@PG32 data]$ psql -d sourcedb
 psql (11.4) 
Type "help" for help. 
sourcedb=# create table logical_tb1(id int primary key,name varchar(20));

/** 在逻辑从库上创建不同的库 */
[postgres@PG33 data]$ createdb desdb 
[postgres@PG33 data]$ psql -d desdb 
psql (11.4) 
Type "help" for help. 
desdb=# create table logical_tb1(id int primary key,name varchar(20)); 
CREATE TABLE 
desdb=# 

/** 在发布主库上创建发布pub1,注意实在sourcedb库下执行 */
 sourcedb=# CREATE PUBLICATION pub1 FOR TABLE logical_tb1; 
CREATE PUBLICATION

desdb=# CREATE SUBSCRIPTION sub1 CONNECTION ‘host=192.168.56.32 port=5432 user=logical_repl dbname=sourcedb password=logical_repl‘ PUBLICATION pub1;
NOTICE:  created replication slot "sub1" on publisher
CREATE SUBSCRIPTION

sourcedb=# SELECT slot_name,plugin,slot_type,database,active,restart_lsn    FROM pg_replication_slots where slot_name=‘sub1‘;
 slot_name |  plugin  | slot_type | database | active | restart_lsn 
-----------+----------+-----------+----------+--------+-------------
 sub1      | pgoutput | logical   | sourcedb | t      | 0/6022D30
(1 row)

desdb=# SELECT * FROM pg_subscription;
 subdbid | subname | subowner | subenabled |                                      subconninfo                                      | subslotname | subsynccommit | subpublications 
---------+---------+----------+------------+---------------------------------------------------------------------------------------+-------------+---------------+-----------------
   24995 | sub1    |       10 | t          | host=192.168.56.32 port=5432 user=logical_repl dbname=sourcedb password=logical_repl | sub1        | off           | {pub1}
(1 row)

sourcedb=# GRANT USAGE ON SCHEMA public TO logical_repl;
GRANT 
sourcedb=# GRANT SELECT ON logical_tb1 TO logical_repl; 
GRANT

/** 在主库插入数据 */ 
sourcedb=# insert into logical_tb1(id,name) values(1,‘a‘),(2,‘bca‘); 
INSERT 0 2

/** 查看数据是否同步完成 */ 
desdb=# select * from logical_tb1;
 id | name 
----+------
  1 | a
  2 | bca
(2 rows)

/** 主库上创建表结构 */ 
sourcedb=# create table logical_tb2(id int primary key ,addr varchar(100)); 
CREATE TABLE 
sourcedb=# 
/** 从库上创建表结构 */
desdb=# create table logical_tb2(id int primary key ,addr varchar(100)); 
CREATE TABLE 
/** 在主库上给逻辑复制账号授权 */ 
sourcedb=# GRANT SELECT ON logical_tb2 TO logical_repl; 
GRANT
/** 添加新表至发布列表 */ 
sourcedb=# ALTER PUBLICATION pub1 ADD TABLE logical_tb2; 
ALTER PUBLICATION 
/** 在主库查看发布列表中的表名 */
sourcedb=# SELECT * FROM pg_publication_tables;
 pubname | schemaname |  tablename  
---------+------------+-------------
 pub1    | public     | logical_tb1
 pub1    | public     | logical_tb2
(2 rows)

/** 主库插入一条记录 */ 
sourcedb=# insert into logical_tb2(id,addr) values(1,‘beijing‘); 
INSERT 0 1 
/** 此时在逻辑从库查看，结果却没有数据 */
desdb=# select  * from logical_tb2;
 id | addr 
----+------
(0 rows)

/** 此时在从库刷新订阅 */ 
desdb=# ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION;
/** 刷新完成后再查数据已经有数据了 */ 
ALTER SUBSCRIPTION
desdb=# select  * from logical_tb2;
 id |  addr   
----+---------
  1 | beijing
(1 row)