FlinkX PostgreSQL WAL实时采集基本原理

{"id":"schema1.test1",
 "schema":"schema1",
"table":"test1",
 "columnList":[
   {"name":"id","type":"int4","index":0},
   {"name":"name","type":"varchar","index":1}
 ],
 "oldData":["2","23"],
 "newData":["2","name1"],
 "type":"UPDATE",
 "currentLsn":23940928,
 "ts":1596358573614
}

CREATE PUBLICATION name
    [ FOR TABLE [ ONLY ] table_name [ * ] [, ...]
      | FOR ALL TABLES ]
    [ WITH ( publication_parameter [= value] [, ... ] ) ]

wal_level 可以选择为minimal, replica, or logical 使用逻辑复制需要设置为logical

fsync boolean类型 表示是否使用fsync()系统调用把WAL文件刷新到物理磁盘，确保数据库在操作系统或硬件奔溃的情况下可恢复到最终状态 默认是on

synchronous_commit boolean类型 声明提交一个事务是否需要等待其把WAL日志写入磁盘后再返回，默认值是’on’

on：默认值，为on且没有开启同步备库的时候,会当wal日志真正刷新到磁盘永久存储后才会返回客户端事务已提交成功,
    当为on且开启了同步备库的时候(设置了synchronous_standby_names),必须要等事务日志刷新到本地磁盘,并且还要等远程备库也提交到磁盘才能返回客户端已经提交.

remote_apply：提交将等待， 直到来自当前同步备用数据库的回复表明它们已收到事务的提交记录并应用它， 以便它对备用数据库上的查询可见。

remote_write：提交将等待，直到来自当前同步的后备服务器的一个回复指示该服务器已经收到了该事务的提交记录并且已经把该记录写出到后备服务器的操作系统。

local：当事务提交时,仅写入本地磁盘即可返回客户端事务提交成功,而不管是否有同步备库。

off：写到缓存中就会向客户端返回提交成功，但也不是一直不刷到磁盘，延迟写入磁盘,延迟的时间为最大3倍的wal_writer_delay参数的(默认200ms)的时间,所有如果即使关闭synchronous_commit,也只会造成最多600ms的事务丢失 可能会造成一些最近已提交的事务丢失，但数据库状态是一致的，就像这些事务已经被干净地中止。但对高并发的小事务系统来说,性能来说提升较大。


wal_sync_method enum类型 用来指定向磁盘强制更新WAL日志数据的方法open_datasync fdatasync fsync_writethrough fsync open_sync



Wal_writer_delay 指定wal writer process 把WAL日志写入磁盘的周期 在每个周期中会先把缓存中的WAL日志刷到磁盘

select * from pg_replication_slots;

Name            Type        References  Description
slot_name       name        复制槽的唯一的集群范围标识符
plugin          name        正在使用的包含逻辑槽输出插件的共享对象的基本名称，对于物理插槽则为null。
slot_type       text        插槽类型 - 物理或逻辑
datoid          oid         该插槽所关联的数据库的OID，或为空。 只有逻辑插槽才具有关联的数据库。
database        text        该插槽所关联的数据库的名称，或为空。 只有逻辑插槽才具有关联的数据库。
active          boolean     如果此插槽当前正在使用，则为真
active_pid      integer     如果当前正在使用插槽，则使用此插槽的会话的进程ID。 NULL如果不活动。
xmin            xid         此插槽需要数据库保留的最早事务。 VACUUM无法删除任何后来的事务删除的元组。
catalog_xmin    xid         影响该插槽需要数据库保留的系统目录的最早的事务。 VACUUM不能删除任何后来的事务删除的目录元组。
restart_lsn     pg_lsn      最老的WAL的地址（LSN）仍然可能是该插槽的使用者所需要的，因此在检查点期间不会被自动移除

wal_level = logical

host replication all 10.0.3.0/24 md5

public static final String PUBLICATION_NAME = "dtstack_flinkx";
public static final String CREATE_PUBLICATION = "CREATE PUBLICATION %s FOR ALL TABLES;";
public static final String QUERY_PUBLICATION = "SELECT COUNT(1) FROM pg_publication WHERE pubname = '%s';";
   
先执行查找sql 判断是否存在 dtstack_flinkx 的 PUBLICATION
如果不存在 执行创建sql语句
conn.createStatement()
    .execute(String.format(CREATE_PUBLICATION, PUBLICATION_NAME));

 ChainedLogicalStreamBuilder builder = conn.getReplicationAPI()
                .replicationStream() //定义一个逻辑复制流
                .logical() //级别是logical
                .withSlotName(format.getSlotName())//复制槽名称
                //协议版本。当前仅支持版本1
                .withSlotOption("proto_version", "1")//槽版本号
                //逗号分隔的要订阅的发布名称列表（接收更改）。 单个发布名称被视为标准对象名称，并可根据需要引用
                .withSlotOption("publication_names", PgWalUtil.PUBLICATION_NAME)//关联的发布名称
                .withStatusInterval(format.getStatusInterval(), TimeUnit.MILLISECONDS);
        long lsn = format.getStartLsn();
        if(lsn != 0){
            builder.withStartPosition(LogSequenceNumber.valueOf(lsn));
        }
        stream = builder.start();

  public void run() {
        LOG.info("PgWalListener start running.....");
        try {
            init();
            while (format.isRunning()) {
                //接收到流对象
                ByteBuffer buffer = stream.readPending();
                if (buffer == null) {
                    continue;
                }
                //解码为table对象 具体信息为库 表 字段信息 WAL id等
                //然后就可以对其进行处理了
                Table table = decoder.decode(buffer);
                if(StringUtils.isBlank(table.getId())){
                    continue;
                }
                String type = table.getType().name().toLowerCase();
                if(!cat.contains(type)){
                    continue;
                }
                if(!tableSet.contains(table.getId())){
                    continue;
                }
                LOG.trace("table = {}",gson.toJson(table));
                ...............
            }
        }
  }