【技术文章】浅淡PostgreSQL流复制

testdb=# SELECT application_name,state FROM pg_stat_replication;
 application_name |   state
------------------+-----------
 standby1         | streaming
 standby2         | streaming
 pg_basebackup    | backup
(3 rows)

synchronous_standby_names = 'standby1'
hot_standby = off
wal_level = archive

testdb=# SELECT application_name AS host,
        write_location AS write_LSN, flush_location AS flush_LSN,
        replay_location AS replay_LSN FROM pg_stat_replication;
   host   | write_lsn | flush_lsn | replay_lsn
----------+-----------+-----------+------------
 standby1 | 0/5000280 | 0/5000280 | 0/5000280
 standby2 | 0/5000280 | 0/5000280 | 0/5000280
(2 rows)

synchronous_standby_names = ''

pg_ctl -D $PGDATA reload

synchronous_standby_names = 'standby1, standby2'

testdb=# SELECT application_name AS host,
         sync_priority, sync_state FROM pg_stat_replication;
   host   | sync_priority | sync_state
----------+---------------+------------
 standby1 |             1 | sync
 standby2 |             2 | potential
(2 rows)

关闭防火墙
systemctl status firewalld.service
systemctl stop firewalld.service
systemctl disable firewalld.service

SELINUX
vi etc/selinux/config
将SELINUX=enforcing 改为SELINUX=disabled

流复制配置
#创建replica用户
create role replica login replication encrypted password 'REPLICA321';

创建物理复制槽
select * from pg_create_physical_replication_slot('node1');
select * from pg_create_physical_replication_slot('node2');
select * from pg_replication_slots;

vi $PGDATA/postgresql.conf
================
listen_addresses = '0.0.0.0' # what IP address(es) to listen on;
port = 5432 # (change requires restart)
max_connections = 1000 # (change requires restart)
superuser_reserved_connections = 13 # (change requires restart)
unix_socket_directories = '.' # comma-separated list of directories
unix_socket_permissions = 0700 # begin with 0 to use octal notation
tcp_keepalives_idle = 60 # TCP_KEEPIDLE, in seconds;
tcp_keepalives_interval = 10 # TCP_KEEPINTVL, in seconds;
tcp_keepalives_count = 10 # TCP_KEEPCNT;
shared_buffers = 2048MB # min 128kB
vacuum_cost_delay = 10 # 0-100 milliseconds
bgwriter_delay = 10ms # 10-10000ms between rounds
wal_level = replica  #minimal replica 逻辑
wal_writer_delay = 10ms # 1-10000 milliseconds
max_wal_senders = 10 # max number of walsender processes
hot_standby = on # "on" allows queries during recovery
wal_receiver_status_interval = 1s # ：send replies at least this often
hot_standby_feedback = on # send info from standby to prevent
log_destination = 'csvlog' # Valid values are combinations of
logging_collector = on # Enable capturing of stderr and csvlog
log_directory = 'pg_log' # directory where log files are written,
log_truncate_on_rotation = on # If on, an existing log file with the
log_rotation_age = 1d # Automatic rotation of logfiles will
log_rotation_size = 10MB # Automatic rotation of logfiles will
log_checkpoints = on
log_connections = on
log_disconnections = on
log_error_verbosity = verbose # terse, default, or verbose messages
log_timezone = 'PRC'
datestyle = 'iso, mdy'
timezone = 'PRC'
lc_messages = 'C' # locale for system error message
lc_monetary = 'C' # locale for monetary formatting
lc_numeric = 'C' # locale for number formatting
lc_time = 'C' # locale for time formatting
full_page_writes = on
synchronous_commit = on # synchronization level;
wal_log_hints = on
synchronous_standby_names = ''
max_replication_slots = 20
archive_mode = on
archive_command = 'cp %p opt/arch/%f && echo %f >> opt/arch/archive.list'

vi postgresql.conf
synchronous_standby_names = 'standby01,standby02‘

recovery_target_timeline='latest'
standby_mode = 'on'
primary_conninfo = 'user=repl password=111111 host=192.168.137.220 port=1921 sslmode=disable sslcompression=1'

standby_mode = 'on'
primary_conninfo = 'application_name=standby02 user=repl password=111111 host=192.168.137.220 port=1921'

Create table t1(id int);
Create table t2(id int);
publication - 发布语法示例
#创建一个发布，发布两个表中所有更改：
CREATE PUBLICATION mypublication FOR TABLE t1, t2;
#创建一个发布，发布所有表中的所有更改：
CREATE PUBLICATION alltables FOR ALL TABLES;
#创建一个发布，只发布一个表中的INSERT操作：
CREATE PUBLICATION insert_only FOR TABLE mydata  WITH (publish = 'insert');
#将发布修改为只发布删除和更新：
ALTER PUBLICATION noinsert SET (publish = 'update, delete');
#给发布添加一些表：
ALTER PUBLICATION mypublication ADD TABLE users, departments;
#查看发布
select * from pg_publication;

CREATE SUBSCRIPTION mysub
         CONNECTION 'host=192.168.200.142 port=5432 user=postgres dbname=postgres'
        PUBLICATION mypublication;
vi postgresql.conf   wal_level = logical

CREATE SUBSCRIPTION mysub
         CONNECTION 'host=192.168.1.50 port=5432 user=foo dbname=foodb'
        PUBLICATION insert_only
               WITH (enabled = false);

ALTER SUBSCRIPTION mysub SET PUBLICATION insert_only;

ALTER SUBSCRIPTION mysub DISABLE;

#postgresql.conf关联配置
wal_level = logical #minimal replica
max_replication_slots = 10        #max_replication_slots 值最少需设置成 1，设置后重启数据库生效。
#创建复制槽
select * from pg_create_logical_replication_slot('log_slot1', 'test_decoding');
#查看复制槽信息             
select * from pg_replication_slots where slot_name='log_slot1';      
#查看复制槽解析记录，记录只能查询一次，之后查询为空，如果想重复查询到日志，
需使用pg_logical_slot_peek_changes() 函数      
SELECT * FROM pg_logical_slot_get_changes('log_slot1', NULL, NULL); 
#使用 pg_recvlogical 接收数据变化
pg_recvlogical -h 127.0.0.1 -d postgres --slot log_slot1 --start -f -

CREATE TABLE parttest (id bigint, test text);
– PRIMARY KEYS don’t work on partitioned tables
CREATE TABLE parttest0 PARTITION OF
parttest FOR VALUES FROM (1) TO (100);
CREATE TABLE parttest1 PARTITION OF
parttest FOR VALUES FROM (101) TO (200);
CREATE PUBLICATION parttest FOR TABLE parttest0,parttest1;

postgresql.conf ：wal_level = logical & listen_addresses = '*'
pg_hba.conf:  host    all repuser  订阅端ip/32     md5

create role repuser login CONNECTION LIMIT 10 replication encrypted password 'repuser';
\du

create table test_lr1(id int4 primary key ,name text);
insert into test_lr1 values (1,'a');
grant connect on database postgres to repuser;
grant usage on schema public to repuser;
grant select on test_lr1 to repuser;
\dp+ test_lr1;

create PUBLICATION pub1 FOR TABLE test_lr1;       #为表test_lr1创建发布
select * from pg_publication;         #查看创建的发布

create table test_lr1(id int4 ,name text);
create subscription sub1 connection 'host=192.168.200.142 port=5432 dbname=postgres user=repuser password=repuser' publication pub1;
select * from pg_subscription;

insert into test_lr1 values (6,'e');

select * from test_lr1 ;

create table test_big2(id int4 primary key, create_time timestamp without time zone default clock_timestamp(), name character varying(32));
grant select on test_big2 to repuser;
ALTER PUBLICATION pub1 add TABLE  test_big2;

create table test_big2(id int4 primary key, create_time timestamp without time zone default clock_timestamp(), name character varying(32));
 ALTER SUBSCRIPTION sub1 REFRESH PUBLICATION ;

insert into test_big2(id,name) select n,n*random()*10000 from generate_series(1,10000000) n ; 
select * from test_big2 ;