文档中心MogDBMogDB StackUqbar
v3.1

文档:v3.1

示例:逻辑复制代码示例

下面示例演示如何通过JDBC接口使用逻辑复制功能的过程。

针对逻辑复制的配置选项,除了在逻辑解码中的配置选项外,还有专门给JDBC流式解码增加的配置项,如下所示:

  1. 解码线程并发度

    通过配置选项parallel-decode-num,指定并行解码的Decoder线程数量。其取值范围为1~20的int型,取1表示按照原有的串行逻辑进行解码,取其余值即为开启并行解码。默认值为1。当该选项配置为1时,禁止配置解码格式decode-style。

  2. 解码格式

    通过配置选项decode-style,指定解码格式。其取值为char型的字符'j'、't'或'b',分别代表json格式、text格式及二进制格式。默认值为'b'即二进制格式解码。该选项仅允许并行解码时设置,且二进制格式解码仅在并行解码场景下支持。

说明: 二进制格式编码规则如下所示:

  1. 前4字节代表接下来到语句级别分隔符字母P(不含)或者该批次结束符F(不含)的解码结果的总字节数,该值如果为0代表本批次解码结束。
  2. 接下来8字节uint64代表lsn。
  3. 接下来1字节的字母有5种B/C/I/U/D,分别代表begin/commit/insert/update/delete。
  4. 第3步字母为B时: a. 接下来的8字节uint64代表CSN。 b. 接下来的8字节uint64代表commit_lsn。 c. 【该部分为可选项】接下来的1字节字母如果为T,则代表后面4字节uint32表示该事务commit时间戳长度,再后面等同于该长度的字符为时间戳字符串。 d. 因为之后仍可能有解码语句,接下来会有1字节字母P或F作为语句间的分隔符,P代表本批次仍有解码的语句,F代表本批次完成。
  5. 第3步字母为C时: a. 【该部分为可选项】接下来1字节字母如果为X,则代表后面的8字节uint64表示xid。 b. 【该部分为可选项】接下来的1字节字母如果为T,则代表后面4字节uint32表示时间戳长度,再后面等同于该长度的字符为时间戳字符串。 c. 因为批量发送日志时,一个COMMIT日志解码之后可能仍有其他事务的解码结果,接下来的1字节字母如果为P则表示该批次仍需解码,如果为F则表示该批次解码结束。
  6. 第3步字母为I/U/D时: a. 接下来的2字节uint16代表schema名的长度。 b. 按照上述长度读取schema名。 c. 接下来的2字节uint16代表table名的长度。 d. 按照上述长度读取table名。 e. 【该部分为可选项】接下来1字符字母如果为N代表为新元组,如果为O代表为旧元组,这里先发送新元组。 i. 接下来的2字节uint16代表该元组需要解码的列数,记为attrnum。 ii. 以下流程重复attrnum次。 1). 接下来2字节uint16代表列名的长度。 2). 按照上述长度读取列名。 3). 接下来4字节uint32代表当前列类型的Oid。 4). 接下来4字节uint32代表当前列的值(以字符串格式存储)的长度,如果为0xFFFFFFFF则表示NULL,如果为0则表示长度为0的字符串。 5). 按照上述长度读取列值。 f. 因为之后仍可能有解码语句,接下来的1字节字母如果为P则表示该批次仍需解码,如果为F则表示该批次解码结束。
  1. 限制仅备机解码

    通过配置选项standby-connection,指定是否限制仅备机解码。其取值为bool型(可用0或1表示),取true(或1)代表限制仅允许连接备机解码,连接主机解码时会报错退出;取false(或0)时不做限制。默认值为false(0)。

  2. 批量发送

    通过配置选项batch-sending,指定是否批量发送。其取值范围为0或1的int型,取0表示逐条发送解码结果,取1表示解码结果累积到达1MB则批量发送解码结果。默认值为0。该选项仅允许并行解码时设置。开启批量发送的场景中,当解码格式为'j'或't'时,在原来的每条解码语句之前会附加一个uint32类型,表示本条解码结果长度(长度不包含当前的uint32类型),以及一个uint64类型,表示当前解码结果对应的lsn。

在并行解码的标准场景下(16核CPU、内存128G、网络带宽 > 200MBps、表的列数为10-100、单行数据量0.1KB~1KB、DML操作以insert为主、不涉及落盘事务即单个事务中语句数量小于4096、parallel_decode_num为8、解码格式为'b'且开启批量发送功能),解码性能(这里以xlog消耗量为标准)不低于100MBps。为保证解码性能达标以及尽量降低对业务影响,一台备机上应尽量仅建立一个并行解码连接,保证CPU、内存、带宽资源充足。

//逻辑复制功能示例:文件名,LogicalReplicationDemo.java
//前提条件:添加JDBC用户机器IP到数据库白名单里,在pg_hba.conf添加以下内容即可:
//假设JDBC用户IP为10.10.10.10
//host    all             all             10.10.10.10/32        sha256
//host    replication     all             10.10.10.10/32        sha256

import org.opengauss.PGProperty;
import org.opengauss.jdbc.PgConnection;
import org.opengauss.replication.LogSequenceNumber;
import org.opengauss.replication.PGReplicationStream;

import java.nio.ByteBuffer;
import java.sql.DriverManager;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class LogicalReplicationDemo {
    public static void main(String[] args) {
        String driver = "org.opengauss.Driver";
 //此处配置数据库IP以及端口,这里的端口为haPort,通常默认是所连接DN的port+1端口
        String sourceURL = "jdbc:opengauss://$ip:$port/postgres";
        PgConnection conn = null;
 //默认逻辑复制槽的名称是:replication_slot
 //测试模式:创建逻辑复制槽
        int TEST_MODE_CREATE_SLOT = 1;
 //测试模式:开启逻辑复制(前提条件是逻辑复制槽已经存在)
        int TEST_MODE_START_REPL = 2;
 //测试模式:删除逻辑复制槽
        int TEST_MODE_DROP_SLOT = 3;
        //开启不同的测试模式
        int testMode = TEST_MODE_START_REPL;

        try {
            Class.forName(driver);
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }

        try {
            Properties properties = new Properties();
            PGProperty.USER.set(properties, "user");
            PGProperty.PASSWORD.set(properties, "passwd");
     //对于逻辑复制,以下三个属性是必须配置项
            PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
            PGProperty.REPLICATION.set(properties, "database");
            PGProperty.PREFER_QUERY_MODE.set(properties, "simple");
            conn = (PgConnection) DriverManager.getConnection(sourceURL, properties);
            System.out.println("connection success!");

            if(testMode == TEST_MODE_CREATE_SLOT){
                conn.getReplicationAPI()
                        .createReplicationSlot()
                        .logical()
                        .withSlotName("replication_slot")
                        .withOutputPlugin("mppdb_decoding")
                        .make();
            }else if(testMode == TEST_MODE_START_REPL) {
                //开启此模式前需要创建复制槽
                LogSequenceNumber waitLSN = LogSequenceNumber.valueOf("6F/E3C53568");
                PGReplicationStream stream = conn
                        .getReplicationAPI()
                        .replicationStream()
                        .logical()
                        .withSlotName("replication_slot")
                        .withSlotOption("include-xids", false)
                        .withSlotOption("skip-empty-xacts", true)
                        .withStartPosition(waitLSN)
                        .withSlotOption("parallel-decode-num", 10) //解码线程并发度
                        .withSlotOption("white-table-list", "public.t1,public.t2") //白名单列表
                        .withSlotOption("standby-connection", true) //强制备机解码
                        .withSlotOption("decode-style", "t") //解码格式
                        .withSlotOption("sending-bacth", 1) //批量发送解码结果
                        .start();
                while (true) {
                    ByteBuffer byteBuffer = stream.readPending();

                    if (byteBuffer == null) {
                        TimeUnit.MILLISECONDS.sleep(10L);
                        continue;
                    }

                    int offset = byteBuffer.arrayOffset();
                    byte[] source = byteBuffer.array();
                    int length = source.length - offset;
                    System.out.println(new String(source, offset, length));

                    //如果需要flush lsn,根据业务实际情况调用以下接口
                    //LogSequenceNumber lastRecv = stream.getLastReceiveLSN();
                    //stream.setFlushedLSN(lastRecv);
                    //stream.forceUpdateStatus();

                }
            }else if(testMode == TEST_MODE_DROP_SLOT){
                conn.getReplicationAPI()
                        .dropReplicationSlot("replication_slot");
            }
        } catch (Exception e) {
            e.printStackTrace();
            return;
        }
    }
}
Copyright © 2011-2024 www.enmotech.com All rights reserved.