9 PostgreSQL 点对点多主表级复制-触发器篇|学习笔记-阿里云开发者社区

开发者学堂课程【PostgreSQL 快速入门：9 PostgreSQL 点对点多主表级复制-触发器篇】学习笔记，与课程紧密联系，让用户快速学习知识。

课程地址：https://developer.aliyun.com/learning/course/16/detail/68

9 PostgreSQL 点对点多主表级复制-触发器篇

内容介绍：

一、表级复制简介

二、操作

三、回顾

一、表级复制简介

前面讲过数据库的集群级的复制，也就是数据库的流复制，相当于是把整个数据库的提取通过 xlog 的一些变更，复制到 standby 节点，sandby 节点跟主数据库保持完全一致。

当然一些不写 xlog 操作无法通过复制给它推广。比如 onload table 创建的像临时表以及 logo 的推广都不会写xlog，所以无法使用流复制。还有比如哈希索引也是不能使用流复制来复制。

表级复制能很的解决上述问题，如果使用流复制复制 standby 节点，它只能作为只读的节点。如果要写，需要激活。

激活之后两个就是重组节点，无法通过 xlog 方法来复制到 standby 节点，也就是激活之后就是两个完全独立的口。

在某一些复制场景，比如只需要复制某一些表或者是某一些表里的数据的，并且复制过去的节点，同时也有读写的操作。在这种情况下，使用流复制无法满足需要，可以使用触发器来完成，或者是使用第三方的插件比如 Slony、londiste3、bucardo、pgpool、pl/proxy。这些插件分为很多种，一种是 sql 分发的方式来控制。比如 pgpool 相当于做中间件，对于要复制的数据，会在它下面配置数据库节点，选择要复制到哪几个数，它相当于是帮你做sql分发的过程。

包括 pl/proxy 也可以作为分发器来实现数据库的复制。然后还有一种是数据库端的触发器的复制。

比如 londiste3以及 slony 都是通过主节点上创建触发器，生成一些操作日志，这些操作日志被作为复制的sql语句发送给 standby 节点，然后在 standby 节点上去执行这些sql语句。

触发器

首先来讲基于触发器的复制，通过 dblink ，postgere 支持 dblink。dplink 整个包的使用可以参考 dblink 的使用手册，就是这篇 blog 里

https：

blog.163.com/digoal@126/blog/static/163877040201321125220134）写通用的触发器来实现multi-master 复制场景。

multi-master 指的是两个 master 节点，比如数据库A数据库B，这两个数据库节点里面的表可以相互进行复制。也就是从A可以把数据复制到B或者从B类也可以把数据复制到A，并且支持异步和同步的复制方法。

对于异步，这种使用方法不是非常严谨，因为异步是同时并发的操作一个表在使用这种触发器，可能会带来一些问题，可能造成在主数据库上执行的顺序和在 standby 节点上执行的事物顺序会有不一致或者其它情况。

但是使用同步就不会有上述问题。尽量使用同步复制，而不要去用异步复制。

二、操作

首先是创建角色，然后要创建 dplink 插件。比如要从30150这一台数据库复制到3.39数据库。这里版本是9.4的版本，然后另外一个是9.3.2版本。

这两个版本之间通过触发器方式可以复制。对于流复制，它就必须要一次，因为这里是两个完全的独立的库。通过输入 create extension dblink 创建 dblink，dblink 已经存在，不需要创建。

另外的3.39库也需要创建，甚至可以从不同的数据库复制到另外的库。只是要使用不同数据的用户，比如专门用于复制的用户叫 test 的用户，可以给它普通的权限。

输入 create test nosuperruser login encrypted password’test’，然后允许它登录，设置密码。

然后在另外数据库。比如在 postgre 数据库也可以创建同名的角色。当然也可以创建不同的角色。

也就是在3.150数据库里面的表要复制到3.39的 postgre，然后通过的用户一样与否关系不大。

在创建用户之后，要赋予一定的权限，包括给它创建 SDM 等，按照这篇 blog 来。

先创建角色。

Postgres =# create role local login encrypted password＇

LOCAL321’;

CREATE ROLE

Postgres =# create role local login encrypted password＇

REMOTR321’;

在另外一个数据库创建远程的角色，然后创建数据库。在3.150数据库创建 local 数据库，在3.39上创建 remote 数据库。

Postgres =# create database local owner local

CEATE DATABASE

postgres =＃create database remote owner remote

CREATE DATABASE

postgres =#\c local postgres

Tou are now connected to database “ local ” as user

“ postgres ”

local=# create extension dblink

CREATE EXTENSION

local=#/c remote postgres

Tou are now connected to database ‘remote’as user postgres

remote = #create extension dblink

然后数据库创建完成之后，先连到 local 下面去创建 local 数据库。3.150上创建 extension，就是 dblink。然后在远程库 remote 的功能也要创建。因为要做 multi-master 的场景，也就是相互都要复制。

数据库创建的报错。因为这是开发版本，可能是开发的一些 bug。换成 pg9.2，这里也是创建角色。

从9.2和9.3之间复制创建用户，创建 local 数据库，然后去创建 dblink，远程的以及创建完成。，接下来在 local 里面创建schema。

local=>/c local local

local=> create schema local

CREATE SCHEMA

local=>/c remote remote

You are now connected to database “remote" as user "remote

remote=> create schema remote

CREATESCHEMA

然后在remote库，也创建schema叫做remote。

接下来要在local库里面去创建测试表，叫做OOC test。本例将使用联合索引

remote=>/c local local

You are now connected to database local" as user “local"

local=>create table loc_test (pkl int, pk2 text, info text,

crt_time times tanp(0), mod_time timestamp(0), primary

ikey(pk1,pk2)):

同时创建表一定要有 pk，用 dblink 去复制的去生成比如本地有一条 sql 语句，在要生成远端执行的 sql 语句，还要调dblink 包里面的函数。

函数要求表必须要有 primary key。也就是被复制的表里面必须要有 primary key，就是有 primary key 能够定位到每一条记录。

local=> select string

agg(positiontext,’’)int2vector,count(*) from

dblink_get_pkey(local loc_test’): string_agg|count

因为使用 dblink 的一些包去生成 sql 语句的，是需要用到比如 primary key 的 numbers 的 enter to the victor 的词，同时还要用到 primary key 的字段。比如是两个字段创建的联合组建的话，这里就是2，然后这是矢量。

比如在 sql 语句当中，select*from loc_test 它的组件的顺序，在整5个字段里面，它是排在1和2的，integer 的值就是1、2。如果组件是排在2、3位的，就是23。

使用 dblink _get_pkey 能够得到是它的 int2vector矢量，还可以得到它总共有多少个组件。

dblink_get_pkey(local loc_test’): string_agg|count

是用于获得一些图像信息的。select*from loc_test 能得到组件信息，它的位置信息以及它的列名。count 是它总共有多少个，组件总共有多少个字段。用于这种通过函数去创建 sql 语句的。

postgres=# \c remote remote

You are now connected to database "remote” as user "renote"

remote=> create table rmt_test (pkl int, ph2 text, info text,

crt_time timestump(0), mod time timestamp 0), prieary key (pkl, pk2)):

NOTICE: CREATE TABLE /PRIMARY KEY will create implicit index

“rat_test pkey" for table "rmt_test" CREATE TABLE

remote=> select string acc(position::text,’’)::int2vector,count(*) fron

dblink_get pkey(remote.rat_test’); string.agg|count

接下来，在远程库里面去创建远程表，表名叫做int_test才是。表结构跟 ioc 一样，只是 schema 的名字和表名不一样。

去查询的，也可以得到它的组件的矢量，组件位置、数量以及组建的包含 primary key 的列的数量。

local 和 remote 库回收 pg_user_rrappings 的 public 权限可以不去做，因为只是把表的查询权限从 public 里面回收掉。如果在 user 里面配密码的话，可以规避一些问题。

回收后要通过 db_link 去连接，

在 local 库创建 server

postgres=# /c local postgre

You are now connected to database "local" as user “postgres"

local=# CREATE FOREIGN DATA WRAPPER pOStresgl VALIDATOR

postgresql_fdw_vali dator:

CREATE FOREIGN DATA WRAPPER

locel=# CREATE SERVER dST FOREIGN DATA WRAPPER pOStgresgl

OPTIONS (hostaddr ’172.16.3.150’, port '9999’, donine

'remote'

options'-ctcp_keepalives_idle=60s-c

tcp_keepalivex_interval=10 -c tcp_keepalives_count=6 -c

application_name=aea_bbb_digosl’): CREATE SERVER

Local=# GRANT USAGE ON FOREIGH SERVER dst TO locel GRANT

local=# CREATE USER MAPPING FOR local SERVER dst OPTIONS

(user’remote’, password ’REMOTE321'): CREATE USER BAPPING

然后是 local 库里面，创建 server叫DST。DST 包含一些连接到远端的信息，比如库名，端口。

39这台监听端口是5432，ip3.39，数据库名是 remote，选项指的是连接从3.150上创建的 server，连接过去的一些选项。

但是注意 application name 一定要配置，并且两个库要配置不一样，因为是使用触发器，才能够规避掉一些无限循环。比如在3.150上面触发触发器，并且生成映射语句，通过 dblink 发送给这台主机。

然后让它去执行，并且这上面同时又有触发器，因为是做多组复制。如果触发器在映射的，它也会触犯 in set，调用远程in set。也就相当于是150 inset 触发39的 inset，然后39被 inset 触发，触发了150 发 inset，就变成无限循环。

所以要通过 application name 来分辨，如果是远程复制过来的 sql，触发器不做任何操作，不会导致无限的插入触发。

接下来就把 server 创建之后，把 server 的权限赋予 local，创建给用户创建 mapper,也就是 server 使用 local 去连接，用 remote 去链接，也就是连接到3.39数据。

配之后，要要来到 remote 库去做配置。remote 库也是同样的道理，不再赘述。在触发器函数在写的，规避掉application name 就可以。注意端口的名字改1924，其它不用改动。然后把权限赋予给 remote 用户。

接下来如果是使用异步复制，会产生一些错误。打个比方，数据库刚不通，那在 local 和 remote 库创建异步复制错误 SQL 记录表。

create table sync_err rec（id serial8 prinary key,nsp_name

name,table_none name, dst_server text, dst query text

create time timestamp without time rone)

MO2wqazTICE CREATE TABLE will create implicil sequence

“sync_err_rec_id_seq" for serial colunn "zync_err_rec. id"

NOTICECREATE TABLE/PRIMARY KEY will create inplicit

index“sync_err_rec pkey for table "sync_err_rec

CREATE TABLE

错误的同步表要连到库里面。

这里面包含几个字段，id字段是组件，nsp_name 也就是 SDV 的名字，表名如果同步错误，同步错误的表面是table_name。目标服务器dst指的是创建的server等等。

技术错误表创建之后，处理错误的函数。处理错误的函数，把它创立在不同的库下面。来看函数里面一些含义。

limit 是错误记录表里面取出多少条数据，比如取出100条，如果当前是 standby 数据库不处理，否则在 server 表里有字段叫 dst_server,用于连接远端数据库的server的。

然后取出来之后，from date 锁住。比如现在正在处理 destination server 的一些记录，不允许当前有其它程序要去处理它。

锁的处是能够使操作是串行的操作。尽量去使用函数同步复制，因为异步复制，如果并行操作，会带来一些事物的执行顺序可能会不一致问题。

比如主节点的事物咨询执行顺序，在多个事物中，有可能是事物B发生的错误，但是事物C正常，最后执行事物的顺序跟在本地执行的顺序不一致。所以尽量要使用一些方法去规避重复问题。当然如果要严格意义上来完全一致，最终还是需要去使用第三方的一些插件比较。

如果服务器取 server 渠道的是空，直接返回，里面没有可以执行的记录，如果是取到，不是空，支持 DST server，再去把里面的一批记录取出来，从错误的 record 表里面去取，然后把dst_query取出来。

同时还要把 id 要记下来，执行完之后要把数据从表里面删掉,因为就比如现在已经同步，一些错误同步记录就可以删掉。

删掉之后，然后就去建立连接。建立连接之前要判断当前会话是否已经建立链接，通过 dblink get connection 就能够得到当前的链接列表，如果包含的 nation name 是指定的取出来的 server 名，就不用重新建立连接。否则就通过dblink connect 命令去建立连接，然后使用 db_link excute 这条命令实施，相当于是去算一些批量的语句，通过聚合函数，还有分号聚合在一起。

使用 true，如果远端执行失败，也直接会回滚，能够确保不会有任何的问题，如果是 false，表示如果远端执行失败，还是继续返回，这是不允许的。在处理错误的，必须要使用true。

接下来就要建触发器。

触发器它的返回类型是 trigger，使用 PUPGCQ 来写的，DST server 变量，是 foreign server 就是要把它复制到哪个server 上去。比如两个节点创建 DST，它的 server 名都叫 DST。

所以在配置的就是都叫 DST。然后连接名不是非常重要，跟 density server 一样就可以。连接状态，连接状态是用dblink connect 函数连接之后产生的状态，通过状态可以去判断连接是否正常。然后触发器的变量是促使用函数可以直接使用的变量，比如 table 的 schema，是 table 的名字。也就是触发触发器的表，它是属于哪个 schema 的。同时它的表名叫什。

目标表的 schema，要复制到其它 schema 的，要制定。v_query，是指使用三个 dblink 的函数得到的sql语句，然后把它放到dblink excute函数里面远程调用，也就相当于是 inset 语句，在触发器里面，通过函数去把它封装成inset语句，然后再通过 dblink excute 去执行 circle。

update，使用 delete 和 insert 去模拟 update。因为使用 dblink 思考，update 无法完整。

比如表叫做 test，然后它的 pk 是 id 和 info 这两个字段，通过 update 指定它的字段的矢量，pk 的字段。矢量是1和2。它总共包含两个字段

圆的 pk 值是1和abc，要把它变更成1和 bcd，变更出来的条件也变成 bcd。一般条件应该是 abc 劳务，最后它等于bcd，所以是 bug，不能使用 update 去封装出远端执行。

这里通过 delete 和 insert 来替代 updata 的操作。因为在 updata 孵化器里面有 new 和 old 的 code。通过 new 和old the record delete 的话，就 delete order,

相当于没有用函数，用函数来得到 update 在远端的执行的 circle。

v_dst_query text 是用于存储 v_query 修改后的语句，主要是修改目标 schema 和目标表名，因为他们可能和原表名不一样。上面三个，是从两个函数里面得到的，所以它封装出来的是 schema 名和表名的话，保持原样，修改 replace 即可。

query update1是 delete 的 circle。然后 query update2是 insert的 circle,下面分别对应是修改后的 delete，是修改的 insert。

然后 v_pk vector 存储 pk 列的逻辑位置。对于9.0以及9.0以上的版本，它的逻辑位置是真正查询出来的位置。对于9.0以下的版本，比如8.4 8.3，它的逻辑列的位置是物理位置。

pk 的字段数是函数要用到的，pk 的名称数组比如 pk1和 pk2，pk 值的数组，是 text 的数组。

临时的 pk 值数组，比如有多个 pk 值，其实是通过聚合把它封装在一起。

excute 状态是调用 dst 的一些返回值。如果使用异步的话，存储返回，如果是同步的话，它就没有意义。

然后接下来是多组复式相关，通过 appliction name 来防止死循环。

客户端 v_application_name_cli text，

v_application_name_check text是要检测的端口，

v_pg_backend_pid int 可以存储 pid，获得 application name。

v_replice_mode text 是复制模式，用于分区分是同步复制还是异步负责。i record 是没有固定格式的一条的数据。y int 用于临时自身的变量。

begin 创建触发器，可以在触发器里面写一些参数。这些参数不是在 function 里面指定，而是在创建触发器指定，通过类似于 ca 的方法来获得的参数传递，不是直接在函数里面写的参数。

它的参数传递是 TG_ARGV的数组。比如从零开始，零就是第一个参数。在创建触发器时，可以在引号内部输入参数值。零号参数是指 foreign server 的名字，也就是触发器复制到 server 里的位置。

v_dst_server 是连接名，然后第二个参数是目标的 schema 的名字，把表复制到远端服务器，远端服务器对应的地方。

v_dst_table_name 是远端的表名。

防止死循环 v_application_name_check，application name 会在连接的时候用到。

同步模式的就是在创建触发器的时候，选择同步复制还是异步复制，同步复制在触发器里面的逻辑是不一样的。接下来的话就是获得当前的 pid，就是行记录触发的 pid，根据 pid，可以获得 application name。

如果是被远程触发过来的链接，带有个 server 属性，server 里面已经抄写 application。比如 server3.150上面连接到3.39，在3.39就能够看到application name，就叫AAABB。同样在3.39连到3.150也会看到它配置的application name。

这里使用的 application name 来确保是远端触发器触发的连接。所以如果它等于提供的 application name，就可以不使用，不会触发无限的循环。

对于9.1以及以前的版本的话，字段叫 prcpid。把 application name 传到客户端，cri 表示是客户端过来的application。然后 application name check 可能是在创建初期要check 的 name。检查是否跟客户端的一致，如果一致，表示是触发器触发的操作，就不会进行循环。

如果是复制程序连上来的会话，如果它相等，返回空，直接返回。相因为连接是远端连接，不是本地的操作。本地人工的操作的 application name 跟 cri 可能不一样，就是当然如果自己也使用一致的 cri，也可以让它 return null, 只要等号两边匹配，就会 return null 防止死循环。

然后是生成一些组件的信息。组件的信息,是从约束里面来获取，约束表里面就有组件的一些信息，比如 tg rr id 是触发器的变量。如果要去看完整的触发器变量，可以在触发器章节，看到它支持的变量。

是在 P2PQCQ里面有章节叫做 trigger，里面的支持的一些变量。

这是在 POPQC 所写的触发器函数里支持的一些函数。比如 tg relid，它对应的就是表的 object，能够获取到它的 pk的值，把长度取出来。长度就是 pk 的字段的个数。比如测试表是 pk1和 pk2，所以它取值2。

接下来获取 pk 的 vector 的信息，以及它的 attname 的数组。attname 的数据，是从 attribute 表里面去取。这里要判断版本。如果是九的版本，就比如9.0或者9.x，下面是8.x。

9.x要的是 dblink 的函数里面，要的是逻辑顺序，比如 note，它要的是逻辑顺序,真正输出的逻辑 code numbers。而以前的版本，它要的是物理顺序，物理顺序就来自 pg_attriute 表里面的个 attnumbers。所以这里有一定的区别。

比如取出来是九的版本，就用窗口函数把它的真正的逻辑顺序取出。

然后如果是八的版本，直接取att 以及 attnumber，把它转换成 int2ector，然后另外一个就直接存在的聚合的值里面就可以。

如果不是8和9的版本，直接就抛出异常。取完之后就要开始判断，如果当前的 operate，也就是当前的触发触发器的是 sql 语句，它是 insert 语句，要生成哪些东西。如果是insert语句，就去从 new 值里面使用 quote literal 函数，相当于是把一些要存储的值封装起来。也就是它会帮你做一些转移。

比如 select quote literal<’\’>，可能有一些需要转移的东西。它有个反斜线符号，它会帮你转移。然后再去使用，可以直接使用这种 quote literal 的值，放在 quote literal execute diving in the execute 里面，就会出现遇到转移的支付报错的问题。所以封装一定使用 quote literal。

相对于使用y去做判断。比如第一次是用一句动态 sql 语句，直接创建数组，然后如果 pk 是多个字段，要往后面再来pad，相当于做值的交换，如果有三个字段不会有问题。只是交换之后，把整个的 pk 拼装成 array_appendto 数据。

上面是 insert，接下来是 delete。

上面数组拼装完之后，使用 build_sql_insert，生成远端执行的 sql 语句，sql 语句就存在 v_carry。v _query 就是调用 d bg execute 要去执行的 sql 语句。

接下来是 delete，如果是 delete，要用到 old。然后再使用 sql delete 去生成 v_carry。

如果触发 sql 的是 update 语句，这里先去做 delete，也就是用 old 的去生成，然后再用 new 去生成 insert 语句。相当于是 update 用到 old 和 new。如果是 truncate 语句，直接 truncate table only 目标表就行。

然后就要把目标 schema 和目标表替换成现有的表名。

比如目标的 schema 和目标的表名，本地表的表名以及 schema 不一样，要做替换操作。比如 local 表名是 local.local test，然后 remote 的表名是 remote.mtn test，所以就有替换的操作。

如果是updata，相当于生成两个 v_query。v_query updata1, v_query updata2 ,一个 delete，一个 resert，都要做replace 的操作，就是去查找目标表名，找到就直接替换成远端的 schema name 和 dst_table_name。

对于 update，因为是两个 sql 语句，所以它们最终在远端执行的 v_dst_query 要合并，就是先执行 delete，然后执行 insert。

否则如果不是 update，就直接替换。

接下来就是去判断如果是异步复制的操作，首先是判断当前表有没有被锁。

也就是前面处理错误日志，有加锁过程。如果正在做远端，做 sql 操作的，对于 server 有操作的，防止顺序错乱，就是事物的执行顺序错乱。只要处理 think error record，写入就直接返回。

如果发现就继续往错误表里面写记录，而不是立刻去调用 dblink excute，能够确保事物的执行顺序的一致性。

然后第二个要判断的是 server 里面是否还有记录。在错误记录里面，目标 server 的复制，还有一些错误没有处理完，不能够直接去调用 dblink excute 去远端执行,也是为确保事物的执行顺序，本地的事物执行顺序和远端的事物执行顺序的一致性。

如果上面判断都没有问题，就表示异步复制可以调用 dblink execute 去复制到远端。

接着判断链接是不是存在。如果不存在，就调用 dblink connect 去产生连接。然后连接完之后，判断是同步复制还是异步复制。如果是同步复制，通过 dblink execute。excute 后面的参数是 true。表示如果远端执行失败，整个的触发器函数都会回滚，本地的 DML 语句也会回滚。

如果是异步复制，这里就是 false，在远端执行失败的话没有关系。远端执行失败，它会返回状态，状态是 error，就往 sql 表里面插入一条记录，sql 语句 v_destination_query，是在远端执行失败的意思，但是不会导致整个本地的事务的回滚，这就是异步复制。

上面是同步复制，如果远端执行失败，本地事务回滚。异步复制如果远端的 circle 记忆失败，本地不需要回滚。

接下来就是异常的捕获，异常的捕获是捕获的任何的上面的一些异常，如果是异步复制，往本地的表里面复制失败的记录，然后抛出 notice，不是警告级。如果是同步复制，捕获到异常之后，就直接抛出错误。表示遇到任何错误，都直接本地回滚。

函数逻辑比较简单，但是比较庞大。

三、回顾

首先还是要在本地库里面，local 库里面再去创建触发器。触发器分为是 insert update 和 delete 的触发器，比如触发器里面的几个参数，目标 servere，是目标 schema 的名字，然后表名是 application name。现在用的是同步复制，也就是如果复制失败，它直接就会报错。在远端也是同样的去创建这两个触发器。

就可以往本地的表里面去写入一些记录。比如在本地表里插入十条结果。在异地的表里面，也能够看到十条记录。在远端同样也插入十条记录，然后在本地也能看到。