逻辑复制

// 认证用的用户名和密码直接写到代码中有很大的安全风险，建议在配置文件或者环境变量中存放(密码应密文存放，使用时解密)，确保安全；
// 本示例以用户名和密码保存在环境变量中为例，运行本示例前请先在本地环境中设置环境变量(环境变量名称请根据自身情况进行设置)EXAMPLE_USERNAME_ENV和EXAMPLE_PASSWORD_ENV。
// $ip、$port、database需要用户自行修改。

import com.huawei.opengauss.jdbc.PGProperty;
import com.huawei.opengauss.jdbc.jdbc.PgConnection;
import com.huawei.opengauss.jdbc.replication.LogSequenceNumber;
import com.huawei.opengauss.jdbc.replication.PGReplicationStream;

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

public class LogicalReplicationDemo {
    private static PgConnection conn = null;
    public static void main(String[] args) {
        String driver = "com.huawei.opengauss.jdbc.Driver";
        //此处配置数据库IP以及端口，这里的端口为haPort，通常默认是所连接DN的port+1端口
        String sourceURL = "jdbc:opengauss://$ip:$port/database";    
        //默认逻辑复制槽的名称是：replication_slot
        //测试模式：创建逻辑复制槽
        int TEST_MODE_CREATE_SLOT = 1;
        //测试模式：开启逻辑复制（前提条件是逻辑复制槽已经存在）
        int TEST_MODE_START_REPL = 2;
        //测试模式：删除逻辑复制槽
        int TEST_MODE_DROP_SLOT = 3;
        //开启不同的测试模式，实际使用时先将testMode设置为TEST_MODE_CREATE_SLOT创建复制槽，
        //testMode设置为TEST_MODE_START_REPL即可开启解码，解码结束后设置为TEST_MODE_DROP_SLOT以删除当前复制槽
        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, System.getenv("EXAMPLE_USERNAME_ENV")); //指定解码用户名
            PGProperty.PASSWORD.set(properties, System.getenv("EXAMPLE_PASSWORD_ENV")); //指定对应密码
            //对于逻辑复制，以下三个属性是必须配置项，如下配置即可
            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") //使用mppdb_decoding插件
                        .make();
            }else if(testMode == TEST_MODE_START_REPL) {
                //开启此模式前需要创建复制槽
                //LogSequenceNumber waitLSN = LogSequenceNumber.valueOf("0/2808340"); //解码的过滤点，在此LSN之前提交的事务将不会被输出
                PGReplicationStream stream = conn
                        .getReplicationAPI()
                        .replicationStream()
                        .logical()
                        .withSlotName("replication_slot") //指定解码的复制槽名称
                        .withSlotOption("include-xids", true) //解码结果中是否包含事务号
                        .withSlotOption("skip-empty-xacts", true) //是否跳过空事务
                        //.withStartPosition(waitLSN) //设置解码的过滤点，在此LSN之前提交的事务将不会被输出
                        .withSlotOption("parallel-decode-num", 10) //解码线程并行度
                        //.withSlotOption("white-table-list", "public.t1,public.t2") //白名单列表
                        //.withSlotOption("standby-connection", true) //强制备机解码
                        .withSlotOption("decode-style", "t") //解码格式
                        //.withSlotOption("sending-batch", 1) //批量发送解码结果
                        .withSlotOption("max-txn-in-memory", 100) //单个解码事务落盘内存阈值为100MB
                        .withSlotOption("max-reorderbuffer-in-memory", 50) //正在处理的解码事务落盘内存阈值为50GB
                        .withSlotOption("exclude-users", "userA") //不返回用户userA执行事务的逻辑日志
                        .withSlotOption("include-user", true) //事务BEGIN逻辑日志携带用户名字
                        .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;
        }
    }
}