PostgreSQL邏輯複製之pglogical篇

時間 2019-11-06

標籤 postgresql 邏輯複製 pglogical 欄目 Postgre SQL 简体版

原文原文鏈接

1、pglogical介紹

pglogical 是 PostgreSQL 的拓展模塊, 爲 PostgreSQL 數據庫提供了邏輯流複製發佈和訂閱的功能。 pglogical 重用了 BDR 項目中的一部分相關技術。pglogical 是一個徹底做爲PostgreSQL 擴展實現的邏輯複製系統。徹底集成，它不須要觸發器或外部程序。這種物理複製的替代方法是使用發佈/訂閱模型複製數據以進行選擇性複製的一種高效方法。支持 PG十、9.六、9.五、9.4 ，提供比 Slony、Bucardo 或 Londiste 更快的複製速度，以及跨版本升級。
咱們使用的下列術語來描述節點和數據流之間的關係，重用了一些早期的 Slony 技術中的術語:git

節點 - PostgreSQL 數據庫實例
發佈者和訂閱者 - 節點的角色名稱
複製集 - 關係表的集合

pglogical 是新技術組件，使用了最新的 PostgreSQL 數據庫中的一些核心功能，因此存在一些數據庫版本限制:github

數據源發佈和訂閱節點須要運行 PostgreSQL 9.4 +
複製源過濾和衝突檢測須要 PostgreSQL 9.5 +

支持的使用場景:sql

主版本數據庫之間的升級(存在上述的版本限制)
完整的數據庫複製
利用複製集，選擇性的篩選的關係表
可從多個上游服務器,作數據的彙集和合並

更多介紹；你們請閱讀
英文版
 中文版數據庫

2、安裝操做

本節介紹了pglogical 擴展模塊複製的基本用法。
下載地址，安裝步驟api

tar -zxvf pglogical-REL2_2_0.tar.gz 
cd pglogical-REL2_2_0
. /home/postgres/.bash_profile
pg_config
USE_PGXS=1 make clean
USE_PGXS=1 make
USE_PGXS=1 make install

首先 PostgreSQL服務器必須正確配置纔可以支持邏輯解碼︰數組

wal_level = 'logical'
# one per database needed on (provider/subscriber)provider node
max_worker_processes = 10  
# one per node needed on provider node
max_replication_slots = 10  
# one per node needed on provider node
max_wal_senders = 10 
shared_preload_libraries = 'pglogical'

若是你想要處理解決與上一次/第一次更新之間的衝突 wins(參閱衝突章節), 你的數據庫版本須要爲PostgreSQL 9.5+ (在9.4中無效) 您能夠向 PostgreSQL.conf 添加此額外的選項:bash

# needed for last/first update wins conflict resolution property available in Postgre
track_commit_timestamp = on

pg_hba.conf 須要配置成容許從本地主機複製，用戶擁有有複製權限,鏈接權限；並重啓數據庫服務服務器

host    replication     postgres        網段ip/24           trust

在全部節點上所對應數據庫安裝pglogical拓展模塊:架構

CREATE EXTENSION pglogical;

3、pglogical複製配置

現有實驗環境

數據庫版本	IP	角色
psql (PostgreSQL) 9.6.0	192.168.1.221	provider
psql (PostgreSQL) 10.5	192.168.1.235	subscriber

3.一、時間同步

服務器時間同步（主備庫都需操做）

echo "*/20 * * * * /usr/sbin/ntpdate -u ntp.api.bz >/dev/null" >> /var/spool/cron/root

3.二、提供者節點配置

一、建立節點

在一個數據庫裏建立提供者節點

# 建立節點
SELECT pglogical.create_node(
    node_name := 'provider1',
    dsn := 'host=192.168.1.221 port=5432 dbname=lottu'
);

二、建立複製集

將public架構中的全部表添加到default複製集中

SELECT pglogical.replication_set_add_all_tables('default', ARRAY['public']);

複製集default的表都必須要primary key

3.三、訂閱者節點配置

一、建立節點

在另外一個數據庫建立訂閱者節點

SELECT pglogical.create_node(
node_name := 'subscriber1',
dsn := 'host=192.168.1.235 port=5432 dbname=lottu'
);

二、建立訂閱

訂閱提供者節點，該訂閱將在後臺啓動同步和複製過程

SELECT pglogical.create_subscription(
subscription_name := 'subscription1',
provider_dsn := 'host=192.168.1.221 port=5432 dbname=lottu'
);

3.四、驗證複製

前面咱們已經完成安裝/配置 pglogical 操做。

一、建立測試表

create table tbl_lottu01(id int primary key, name text, reg_time timestamp);

因爲須要驗證insert/update/delete/truncate操做是否同步；因此建立的表要有主鍵。固然只對發佈者必需要主鍵約束。

二、添加測試數據

lottu=# insert into tbl_lottu01 select generate_series(1,10000),'lottu',now();
INSERT 0 10000

三、將表添加對應的複製集

對新建的表；並無爲其分配對應的複製集；須要手動添加。固然能夠利用觸發器自動添加；後續補充。

lottu=# select * from pglogical.replication_set_table ;
 set_id | set_reloid | set_att_list | set_row_filter 
--------+------------+--------------+----------------
(0 rows)

方法1：

前面講解建立複製集中；3.2.2中「將public架構中的全部表添加到default複製集中」

SELECT pglogical.replication_set_add_all_tables('default', ARRAY['public']);

方法二：

將表添加到對應的複製集中；詳細介紹能夠查看前面文檔。

pglogical.replication_set_add_table(set_name name, relation regclass, synchronize_data boolean, columns text []，row_filter text)

兩種方法均可以；咱們採用第二種方法。

lottu=# select pglogical.replication_set_add_table( set_name := 'default', relation := 'tbl_lottu01',synchronize_data := true);
 replication_set_add_table 
---------------------------
 t
(1 row)

咱們查看複製集

lottu=# select * from pglogical.replication_set_table ;
  set_id   | set_reloid  | set_att_list | set_row_filter 
-----------+-------------+--------------+----------------
 290045701 | tbl_lottu01 |              | 
(1 row)

同時，數據也同步到 subscriber 節點。由於在第二種方法有同步的操做。若使用第一種方法；還須要在subscriber 節點同步表的操做。

#從新同步一個表
pglogical.alter_subscription_resynchronize_table(subscription_name name, relation regclass) 
#將全部的表都同步
pglogical.alter_subscription_synchronize(subscription_name name, truncate bool)

四、查看subscriber 節點

查看錶 tbl_lottu01 信息

lottu=# select * from pglogical.show_subscription_table('subscription1','tbl_lottu01');
 nspname |   relname   |    status    
---------+-------------+--------------
 public  | tbl_lottu01 | synchronized
(1 row)

lottu=# select count(1) from tbl_lottu01;
 count 
-------
 10000
(1 row)

在複製集default中： update/delete/truncate 操做也是同步複製。不做演示

複製集	INSERT	UPDATE	DELETE	TRUNCATE
default	√	√	√	√
default_insert_only	√	×	×	×

4、複製特性擴展

4.一、延遲複製

pglogical.create_subscription(subscription_name name, provider_dsn text, replication_sets text[], synchronize_structure boolean, synchronize_data boolean, forward_origins text[], apply_delay interval)

參數:

subscription_name - 訂閱的名稱，必須是惟一的
provider_dsn - 提供者的鏈接字符串
replication_sets - 要訂閱的複製集數組，這些必須已存在，默認爲「{default，default_insert_only，ddl_sql}」
synchronize_structure - 指定是否將提供者與訂閱者之間的結構同步，默認爲false
synchronize_data - 指定是否將數據從提供者同步到訂閱者，默認爲true
forward_origins - 要轉發的原始名稱數組，當前只支持的值是空數組，意味着不轉發任何不是源自提供者節點的更改，或「{all}」這意味着複製全部更改，不管它們的來源是什麼，默認是所有}」
apply_delay - 延遲複製多少，默認爲0秒

示例：數據表結構同步；且延遲複製1分鐘

SELECT pglogical.create_subscription(
subscription_name := 'subscription1',
provider_dsn := 'host=192.168.1.221 port=5432 dbname=lottu',
synchronize_structure := true,
apply_delay := '00:01:00'::interval
);

4.二、對源端進行行/列過濾

過濾機制須要 PostgreSQL 9.5 +

pglogical.replication_set_add_table(set_name name, relation regclass, synchronize_data boolean, columns text []，row_filter text)

參數：

set_name - 現有複製集的名稱
relation - 要添加到集合中的表的名稱或OID
synchronize_data - 若是爲true，則表數據將在訂閱給定複製集的全部訂戶上同步，默認爲false
columns - 要複製的列的列表。一般，當應複製全部列時，這將設置爲NULL，這是默認值
row_filter - 行過濾表達式，默認爲NULL（無過濾），有關詳細信息，請參閱（行過濾）。警告：在使用有效行篩選器同步數據時要當心。使用synchronize_data=true有效row_filter就像對錶的一次性操做。使用修改後再次執行它將row_filter不會將數據同步到訂戶。訂閱者可能須要pglogical.alter_subscription_resynchronize_table()來修復它。

示例：對錶tbl_lottu02中字段{id, name, job} 字段列過濾；且對條件 ‘id > 10’ 進行行過濾 **

# provider 節點 建立表並插入測試數據
create table tbl_lottu02 (id int primary key, name text, job text, reg_time timestamp );
insert into tbl_lottu02 select generate_series(1,20) id,'lottu'||generate_series(1,20),'pg', now();

# subscriber節點建立表； 能夠只建立複製的列的數據表
create table tbl_lottu02 (id int primary key, name text, job text, reg_time timestamp );
# or
create table tbl_lottu02 (id int primary key, name text, job text);

#provider 節點 將表加入複製集中；並同步記錄
lottu=# select pglogical.replication_set_add_table(set_name := 'default', relation := 'tbl_lottu02', synchronize_data := true, columns := '{id, name, job}',row_filter := 'id < 10');
 replication_set_add_table 
---------------------------
 t
(1 row)

# subscriber節點查看錶tbl_lottu02記錄
lottu=# select * from tbl_lottu02;
 id |  name  | job 
----+--------+-----
  1 | lottu1 | pg
  2 | lottu2 | pg
  3 | lottu3 | pg
  4 | lottu4 | pg
  5 | lottu5 | pg
  6 | lottu6 | pg
  7 | lottu7 | pg
  8 | lottu8 | pg
  9 | lottu9 | pg
(9 rows)

4.三、爲新表自動分配複製集

事件觸發器工具可用於描述爲新建立的表定義複製集的規則。

CREATE OR REPLACE FUNCTION pglogical_assign_repset()
RETURNS event_trigger AS $$
DECLARE obj record;
BEGIN
    FOR obj IN SELECT * FROM pg_event_trigger_ddl_commands()
    LOOP
        IF obj.object_type = 'table' THEN
            IF obj.schema_name = 'config' THEN
                PERFORM pglogical.replication_set_add_table('configuration', obj.objid);
            ELSIF NOT obj.in_extension THEN
                PERFORM pglogical.replication_set_add_table('default', obj.objid);
            END IF;
        END IF;
    END LOOP;
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER pglogical_assign_repset_trg
    ON ddl_command_end
    WHEN TAG IN ('CREATE TABLE', 'CREATE TABLE AS')
    EXECUTE PROCEDURE pglogical_assign_repset();

4.四、衝突檢測

衝突檢測須要 PostgreSQL 9.5 +
若是節點訂閱多個提供程序，或當本地寫入在訂閱服務器上發生，可能會發生衝突，尤爲是對傳入的變化。這些都自動檢測，並能夠就此採起行動取決於配置。
解決衝突的辦法是經過配置 pglogical.conflict_resolution 參數。
pglogical.conflict_resolution 支持的配置參數選項爲︰

error - 複製將中止上錯誤若是檢測到衝突和手動操做須要解決
apply_remote - 老是應用與本地數據有衝突的更改，這是默認值
keep_local - 保留數據的本地版本，並忽略來自遠程節點相互衝突的更改
last_update_wins - 時間戳爲提交最新的版本(newest commit timestamp)的數據將會被保存(這能夠是本地或遠程版本）
first_update_wins - 時間戳爲最舊的版本(oldest timestamp)的數據將會被保存(這能夠是本地或遠程版本）

當參數track_commit_timestamp被禁用時，惟一容許的配置值是 apply_remote。 PostgreSQL 9.4 不支持 track_commit_timestamp 配置參數只能配置參數apply_remote(該參數是默認值)。

# 在 訂閱者 節點配置;咱們保留最新的數據
track_commit_timestamp = on
pglogical.conflict_resolution = 'last_update_wins'

# 在 訂閱者 節點建立測試表tbl_lottu03
lottu=# create table tbl_lottu03(id int primary key, name text);
CREATE TABLE
lottu=# insert into tbl_lottu03 values (1001,'subscriber');
INSERT 0 1

# 在 發佈者 節點 建立測試表
create table tbl_lottu03(id int primary key, name text);
select pglogical.replication_set_add_table( set_name := 'default', relation := 'tbl_lottu03',synchronize_data := true);
insert into tbl_lottu03 values (1001,'provider');

# 在 訂閱者 節點 查看數據
lottu=# select * from tbl_lottu03;
  id  |   name   
------+----------
 1001 | provider

後記：在訂閱者的表須要主鍵約束；否則檢測不到衝突；是否須要主鍵約束固然這個也是根據需求而定。

5、場景介紹

5.一、可從多個上游服務器,作數據的彙集和合並

發佈者跟訂閱者的關係；一個發佈者能夠被多個訂閱者訂閱。多個發佈者能夠被同一個訂閱者訂閱。

數據庫版本	IP	數據庫	角色
psql (PostgreSQL) 9.6.0	192.168.1.221	lottu	provider1
psql (PostgreSQL) 9.6.0	192.168.1.221	lottu02	provider2
psql (PostgreSQL) 10.5	192.168.1.235	lottu	subscriber

爲了加以區分；咱們定製SQL提示符；例如

lottu=# \set PROMPT1 '%`echo provider1=`'
provider1=

5.1.一、建立測試表

# 每一個節點建立測試表; 訂閱者建立的表能夠無主鍵；若訂閱者有主鍵，可利用序列自增來解決衝突。(例如：本例是兩個發佈者，則發佈者1可取奇數;發佈者二可取偶數)。若無主鍵;數據不受影響。
provider1=create table tbl_lottu05(id int primary key,name text);
CREATE TABLE
provider1=CREATE SEQUENCE seq_lottu05_id INCREMENT BY 2 START WITH 1;
CREATE SEQUENCE

provider2=create table tbl_lottu05(id int primary key,name text);
CREATE TABLE
provider2=CREATE SEQUENCE seq_lottu05_id INCREMENT BY 2 START WITH 2;
CREATE SEQUENCE

subscriber=create table tbl_lottu05(id int primary key,name text);
CREATE TABLE

5.1.二、搭建模擬場景

更多介紹查看第三節；或者查考《PostgreSQL 邏輯複製文檔 (pglogical 文檔 )》

# provider 節點1
provider1=SELECT pglogical.create_node(node_name := 'provider1', dsn := 'host=192.168.1.221 port=5432 dbname=lottu');
 create_node 
-------------
  2976894835
  
provider1=select pglogical.replication_set_add_table( set_name := 'default', relation := 'tbl_lottu05',synchronize_data := true);
 replication_set_add_table 
---------------------------
 t

# provider 節點2
provider2=SELECT pglogical.create_node(node_name := 'provider2', dsn := 'host=192.168.1.221 port=5432 dbname=lottu02');
 create_node 
-------------
  1828187473

provider2=select pglogical.replication_set_add_table( set_name := 'default', relation := 'tbl_lottu05',synchronize_data := true);
 replication_set_add_table 
---------------------------
 t

# subscriber 節點
subscriber=SELECT pglogical.create_node(node_name := 'subscriber', dsn := 'host=192.168.1.235 port=5432 dbname=lottu');
 create_node 
-------------
  2941155235
  
subscriber=SELECT pglogical.create_subscription(subscription_name := 'subscription1', provider_dsn := 'host=192.168.1.221 port=5432 dbname=lottu');
 create_subscription 
---------------------
          1763399739
        
subscriber=SELECT pglogical.create_subscription(subscription_name := 'subscription2', provider_dsn := 'host=192.168.1.221 port=5432 dbname=lottu02'); 
create_subscription 
---------------------
          1871150101

5.1.三、插入數據驗證

provider1=insert into tbl_lottu05 select nextval('seq_lottu05_id'),'lottu' || generate_series(1,10,2);
INSERT 0 5

provider2=insert into tbl_lottu05 select nextval('seq_lottu05_id'),'lottu' || generate_series(1,10,2);
INSERT 0 5

subscriber=select * from tbl_lottu05;
 id |  name  
----+--------
  1 | lottu1
  3 | lottu3
  5 | lottu5
  7 | lottu7
  9 | lottu9
  2 | lottu1
  4 | lottu3
  6 | lottu5
  8 | lottu7
 10 | lottu9
(10 rows)

5.二、數據庫版本升級

pglogical 對 PostgreSQL 版本升級是一個很實用的工具。能實現以幾乎爲零的停機時間遷移和升級PostgreSQL。侷限性在於pglogical支持的 PostgreSQL 版本。
本例簡單模擬下pglogical 對 PostgreSQL 版本升級；忽略插件、存儲空間、表空間、以及業務SQL和自定義函數建立。

數據庫版本	IP	數據庫	角色
psql (PostgreSQL) 9.6.0	192.168.1.221	lottu	provider
psql (PostgreSQL) 10.5	192.168.1.235	lottu	subscriber

5.2.一、新建升級數據庫

以一個全新的數據庫進行操做

PG10-235=drop database if exists  lottu;
NOTICE:  database "lottu" does not exist, skipping
DROP DATABASE
PG10-235=create database lottu owner lottu;
CREATE DATABASE

5.2.二、pglogical 插件安裝

本環境已經安裝pglogical；只要到對應數據庫建立pglogical插件便可

PG10-235=CREATE EXTENSION pglogical;
CREATE EXTENSION
PG10-235=\dx
                   List of installed extensions
   Name    | Version |   Schema   |          Description           
-----------+---------+------------+--------------------------------
 pglogical | 2.2.0   | pglogical  | PostgreSQL Logical Replication
 plpgsql   | 1.0     | pg_catalog | PL/pgSQL procedural language
(2 rows)

5.2.三、配置pglogical

發佈者節點

這個要根據真實環境來設置；考慮到真實環境數據庫中表不必定都有主鍵約束，可將表放到複製集 "default_insert_only"。

PG96-221=SELECT pglogical.create_node(node_name := 'provider', dsn := 'host=192.168.1.221 port=5432 dbname=lottu');
 create_node 
-------------
  3171898924
(1 row)

PG96-221=SELECT pglogical.replication_set_add_all_tables('default_insert_only', ARRAY['public']);
 replication_set_add_all_tables 
--------------------------------
 t
(1 row)

該函數可實現主鍵和非主鍵分別放到'default'和'default_insert_only'複製集

CREATE OR REPLACE FUNCTION "public"."pglogical_relhaspkey_repset"()
  RETURNS "pg_catalog"."void" AS $BODY$ DECLARE obj record;
BEGIN
  FOR obj IN (SELECT n.nspname, c.relname, c.relhaspkey
                FROM pg_catalog.pg_class c
                LEFT JOIN pg_catalog.pg_namespace n
                  ON n.oid = c.relnamespace
               WHERE c.relkind = 'r'
                 AND n.nspname <> 'pg_catalog'
                 AND n.nspname <> 'information_schema'
                 AND n.nspname !~ '^pg_toast'
                 AND pg_catalog.pg_table_is_visible(c.oid)
               ORDER BY 1, 2) LOOP
      IF obj.relhaspkey THEN
        PERFORM pglogical.replication_set_add_table(set_name := 'default', relation := obj.relname :: regclass);
      ELSE
        PERFORM pglogical.replication_set_add_table(set_name := 'default_insert_only', relation :=  obj.relname :: regclass);
      END IF;
  END LOOP;
END; $BODY$
  LANGUAGE plpgsql VOLATILE
  COST 100

訂閱者節點

PG10-235=SELECT pglogical.create_node(node_name := 'subscriber', dsn := 'host=192.168.1.235 port=5432 dbname=lottu');
 create_node 
-------------
  2941155235

5.2.四、遷移DDL

pglogical 能夠同步表/序列結構；在建立訂閱者 'pglogical.create_subscription' ; 裏面參數synchronize_structure - 指定是否將提供者與訂閱者之間的結構同步，默認爲false。能夠同步表/序列/索引。

PG10-235=SELECT pglogical.create_subscription(subscription_name := 'subscription', provider_dsn := 'host=192.168.1.221 port=5432 dbname=lottu', synchronize_structure := true, synchronize_data := false);
 create_subscription 
---------------------
          2875150205
(1 row)

5.2.五、業務代碼改寫優化

上一步咱們沒同步數據。因此參數synchronize_data咱們選擇false。雖然把表/序列/索引結構同步過來；可是業務代碼(函數/插件)沒同步過來；還要考慮這些業務代碼是否須要改寫優化。由於新的版本每每有新特性。

5.2.六、全量複製

pglogical有將全部未同步表都在單個操做中同步
語法：

pglogical.alter_subscription_synchronize(subscription_name name, truncate bool)

參數：

subscription_name - 現有訂閱的名稱
truncate - 若是爲true，表將在複製前被截斷，默認爲false

PG10-235=SELECT pglogical.alter_subscription_synchronize(subscription_name := 'subscription', truncate := false);
 alter_subscription_synchronize 
--------------------------------
 t
(1 row)

5.2.七、比對數據一致

通過上一步，兩個數據庫數據達到一致。

查看錶同步狀態

PG10-235=select * from pglogical.show_subscription_table(subscription_name := 'subscription', relation := 'tbl_lottu01'::regclass);
 nspname |   relname   |    status    
---------+-------------+--------------
 public  | tbl_lottu01 | synchronized
(1 row)

比對兩個數據庫表的數據

PG96-221=select count(1) from tbl_lottu01;
 count 
-------
 10000
(1 row)

PG10-235=select count(1) from tbl_lottu01;
 count 
-------
 10000
(1 row)

5.2.八、業務切換

比對數據一致；能夠將業務切換到升級後的數據庫。

5.2.九、刪除pglogical配置

這步是可選的；保證升級後的數據庫正常支持業務。不存在數據丟失的狀況下。能夠刪除pglogical配置。
刪除步驟：

刪除訂閱信息
刪除兩個數據庫pglogical節點

PG10-235=select pglogical.drop_subscription(subscription_name := 'subscription',ifexists := true);
 drop_subscription 
-------------------
                 1
(1 row)

PG10-235=select pglogical.drop_node(node_name := 'subscriber', ifexists := true);
 drop_node 
-----------
 t
(1 row)

PG96-221=select pglogical.drop_node(node_name := 'provider', ifexists := true);
 drop_node 
-----------
 t
(1 row)