# 逻辑复制
本节示例演示如何通过JDBC接口使用逻辑复制功能。针对逻辑复制的配置选项，除了参考《特性指南》的"逻辑复制 \> 逻辑解码"章节中的配置选项外，还有专门给JDBC等流式解码工具增加的配置项，如下所示：
1. 解码线程并行度 通过配置选项parallel-decode-num，指定并行解码的Decoder线程数量。取值类型为int型，取值范围为1\~20，取1表示按照原有的串行逻辑进行解码，取其余值即为开启并行解码。默认值为1。当该选项配置为1时，禁止配置以下选项：解码格式选项decode-style、批量发送选项sending-batch和并行解码队列长度选项parallel-queue-size。
   
2. 解码格式
   通过配置选项decode-style，指定解码格式。其取值为char型的字符'j'、't'或'b'，分别代表json格式、text格式及二进制格式。该选项仅允许并行解码时设置，且二进制格式解码仅在并行解码场景下支持。对于json和text格式解码，在批量发送的解码结果中，每条解码语句的前4字节组成的uint32代表该条语句总字节数（不包含该uint32类型占用的4字节，0代表本批次解码结束），8字节uint64代表相应lsn（begin对应first_lsn，commit对应end_lsn，其他场景对应该条语句的lsn）。
   ![](https://support.huaweicloud.com/distributed-devg-v8-gaussdb/public_sys-resources/note_3.0-zh-cn.png)
   二进制格式编码规则如下所示：
   1. 前4字节代表接下来到语句级别分隔符字母P（不含）或者该批次结束符F（不含）的解码结果的总字节数，该值如果为0代表本批次解码结束。
   
   2. 接下来8字节uint64代表相应lsn（begin对应first_lsn，commit对应end_lsn，其他场景对应该条语句的lsn）。
   
   3. 接下来1字节的字母有5种B/C/I/U/D，分别代表begin/commit/insert/update/delete。
   
   4. 当第3步字母为B时：
      1. 接下来的8字节uint64代表CSN。
      
      2. 再接下来的8字节uint64代表first_lsn。
      
      3. 【可选】接下来的1字节字母如果为T，则代表后面4字节uint32表示该事务commit时间戳长度，再后面等同于该长度的字符为时间戳字符串。
      
      4. 【可选】接下来的1字节字母如果为N，则代表后面4字节uint32表示该事务用户名的长度，再后面等同于该长度的字符为事务的用户名字。
      
      5. 之后仍可能有解码语句，接下来会有1字节字母P或F作为语句间的分隔符，P代表本批次仍有解码的语句，F代表本批次解码完成。
       
   
   5. 当第3步字母为C时：
      1. 【可选】接下来1字节字母如果为X，则代表后面的8字节uint64表示xid。
      
      2. 【可选】接下来1字节字母如果为T，则代表后面的4字节uint32表示时间戳长度，再后面等同于该长度的字符为时间戳字符串。
      
      3. 因为批量发送日志时，一个COMMIT日志解码之后可能仍有其他事务的解码结果，接下来的1字节字母如果为P则表示该批次仍需解码，如果为F则表示该批次解码结束。
       
   
   6. 当第3步字母为I/U/D时：
      1. 接下来的2字节uint16代表schema名的长度。
      
      2. 按照上述长度读取schema名。
      
      3. 接下来的2字节uint16代表table名的长度。
      
      4. 按照上述长度读取table名。
      
      5. 【可选】接下来1字节字母如果为N代表为新元组，如果为O代表为旧元组，这里先发送新元组。
         1. 接下来的2字节uint16代表该元组需要解码的列数，记为attrnum。
         
         2. 以下流程重复attrnum次。
            1. 接下来2字节uint16代表列名的长度。
            
            2. 按照上述长度读取列名。
            
            3. 接下来4字节uint32代表当前列类型的OID。
            
            4. 接下来4字节uint32代表当前列的值（以字符串格式存储）的长度，如果为0xFFFFFFFF则表示NULL，如果为0则表示长度为0的字符串。
            
            5. 按照上述长度读取列值。
             
          
      
      6. 因为之后仍可能有解码语句，接下来的1字节字母如果为P则表示该批次仍需解码，如果为F则表示该批次解码结束。
       
     
3. 限制仅备机解码 通过配置选项standby-connection，指定是否限制仅备机解码。其取值为Boolean型（可用0或1表示），取true（或1）代表限制仅允许连接备机解码，连接主机解码时会报错退出。取false（或0）时不做限制。默认值为false（0）。
   
4. 批量发送 通过配置选项sending-batch，指定是否批量发送。其取值范围为int型的0或1，取0表示逐条发送解码结果，取1表示解码结果累积到达1MB则批量发送解码结果。默认值为0。该选项仅允许并行解码时设置。开启批量发送的场景中，当解码格式为'j'或't'时，在原来的每条解码语句之前会附加一个uint32类型，表示本条解码结果长度（长度不包含当前的uint32类型），以及一个uint64类型，表示当前解码结果对应的lsn。
   
5. 并行解码队列长度 通过配置选项parallel-queue-size，指定并行逻辑解码线程间进行交互的队列长度。取值范围\[2，1024\]，且必须为2的幂数，默认值为128。队列长度和解码过程的内存使用量正相关。
   
6. 逻辑解码内存阈值 逻辑解码内存阈值通过配置选项max-txn-in-memory指定单个事务解码中间结果缓存的内存阈值，单位为MB。并行解码模式下，该参数已废弃，不生效。串行解码模式下，取值范围为\[0，100\]，默认值为0，表示不管控内存使用。
   通过配置选项max-reorderbuffer-in-memory指定所有事务解码中间结果缓存的内存阈值，单位为GB。串行解码模式下，取值范围：\[0，100\]，默认值为0、表示不管控内存使用。并行解码模式下，取值范围：\[1，max_process_memory总量的50%\]，默认值为1与max_process_memory/1048576\*10%的较大值，其中1048576为kB到GB的单位转换。
   当超过内存阈值时，解码过程将出现解码中间结果写临时文件的现象，影响逻辑解码的性能。临时文件的大小与事务修改的数据量成正比。如果临时文件过多，可能会导致磁盘进入只读状态的风险。
   
7. 逻辑解码表元信息内存阈值 逻辑解码表元信息通过配置选项desc-memory-limit指定逻辑解码任务维护的所有表元信息内存开销的上限，单位为MB，取值范围为\[10, 1024\]，默认值为100。解码过程中，维护的表元信息总内存超过该阈值时，会触发表元信息的FIFO内存淘汰机制，清理部分内存。
   
8. 逻辑解码发送超时阈值 通过配置选项sender-timeout指定内核与客户端的心跳超时阈值。当该时间段内没有收到客户端任何消息，逻辑解码将主动停止，并断开和客户端的连接。单位为毫秒（ms），取值范围为\[0, 2147483647\]，默认值取决于GUC参数logical_sender_timeout配置。设置为0，表示逻辑解码不会主动断开和客户端的连接；如果设置过小，例如1ms，则可能存在解码任务中断的风险。
   
9. 逻辑解码用户黑名单选项 使用逻辑解码用户黑名单，逻辑解码结果将过滤黑名单中用户的事务操作。当前相关选项如下：
   1. exclude-userids：指定黑名单用户的OID，多个OID通过逗号分隔，不校验用户OID是否存在。同一个业务用户在不同DN的OID不一定相同。当前分布式直连DN逻辑解码需要传入业务用户在各DN上相应的OID，否则可能出现某些DN逻辑解码结果进行了过滤而某些DN节点未过滤的情况。
   
   2. exclude-users：指定黑名单用户名称，多个名字通过逗号分隔，通过dynamic-resolution设置是否动态解析识别用户名称。若解码报错用户不存在而出现中断，在确定日志产生时刻不存在对应的黑名单用户，可以通过配置dynamic-resolution为true或者从用户黑名单中删除报错用户名称来启动解码继续获取逻辑日志。
   
   3. dynamic-resolution：是否动态解析黑名单用户名称，默认值为true。设置为false时，当解码检测到黑名单exclude-users中用户不存在时，将报错并退出逻辑解码。设置为true时，当解码检测到黑名单exclude-users中用户不存在时继续解码。
    
10. 事务逻辑日志输出选项
    1. include-xids：事务的BEGIN逻辑日志是否输出事务ID，默认值为true。
    
    2. include-timestamp：事务的BEGIN逻辑日志是否输出事务提交时间，默认值为false。
    
    3. include-user：事务的BEGIN逻辑日志是否输出事务的用户名字，默认值为false。事务的用户名字特指授权用户（执行事务对应会话的登录用户），它在事务的整个执行过程中不会发生变化。
     
11. JDBC默认设置逻辑解码连接的socketTimeout=10s，备机解码在主机压力大的时候可能会导致连接超时关闭，可以通过配置withStatusInterval(10000,TimeUnit.MILLISECONDS)，调整超时时间来解决连接超时关闭的问题。
12. 心跳日志输出选项 enable-heartbeat：是否输出心跳日志，默认值为false。
    ![](https://support.huaweicloud.com/distributed-devg-v8-gaussdb/public_sys-resources/note_3.0-zh-cn.png)
    以下对心跳日志进行解析，具体格式见下图（考虑到前向兼容性，相关部分仍保留着LSN的命名方式，实际含义依具体场景而定）：
    - 二进制格式首先是字符'h'，表示消息是心跳日志。
    
    - 之后是心跳日志内容。第一个8字节uint64，直连DN解码场景代表LSN，表示发送心跳逻辑日志时读取的WAL日志结束位置；第二个8字节uint64，直连DN解码场景代表LSN，表示发送心跳逻辑日志时刻已经落盘的WAL日志的位置；第三个8字节uint64代表时间戳（从1970年1月1日开始），表示最新解码到的事务日志或检查点日志的产生时间戳。
    
    - 关于消息结束符：如果是二进制格式，则为字符'F'。如果格式为text或者json且为批量发送则结束符为'0'，否则没有结束符。消息内容采用大端字节序进行数据传输。
    
    
    ![](https://support.huaweicloud.com/distributed-devg-v8-gaussdb/figure/zh-cn_image_0000002526898382.png "点击放大")
    
13. 逻辑解码控制参数，用于控制DDL的反解析流程以及输出形式：
    1. enable-ddl-decoding：默认false，不开启DDL语句的逻辑解码。值为true时，开启DDL语句的逻辑解码。
    
    2. enable-ddl-json-format：默认false，传送TEXT格式的DDL反解析结果。值为true时，传送JSON格式的DDL反解析结果。
     
#### 示例
![](https://support.huaweicloud.com/distributed-devg-v8-gaussdb/public_sys-resources/notice_3.0-zh-cn.png)
- 逻辑复制类PGReplicationStream为非线程安全类，禁止多线程使用同一个PGReplicationStream对象。
- PGReplicationStream创建和使用的逻辑复制槽名称仅支持小写字母、数字以及"_"，长度最大为63。
- 在直连DN解码场景下，需要保证客户端与目标DN节点之间的网络连通正常。
  
- 在直连DN解码场景下，为了收集所有DN分片上的数据，需要对所有DN分片节点分别建立连接并执行解码。每个DN分片节点都需要配置一个逻辑复制槽（建议每个DN分片创建同名的逻辑复制槽），以启动对应的解码任务。
 
代码运行的前提条件：
1. 根据实际情况添加gaussdbjdbc.jar包（例如用户使用IDE执行代码，则需要在本地IDE添加gaussdbjdbc.jar包）。
2. 添加JDBC用户机器IP（假设IP为10.11.12.34）到复制数据权限的白名单里，命令如下：
   ```
   直连DN解码：
   gs_guc reload -Z datanode -N all -I all -h 'host replication all 10.11.12.34/32 sha256'
   ```
   
3. 将wal_level参数设置为logical，设置方法请联系管理员处理。
4. 创建表t1和t2，并且对该表进行DDL或DML操作。以下是单个DN分片的解码任务程序示例。
 
```
// 认证用的用户名和密码直接写到代码中有很大的安全风险，建议在配置文件或者环境变量中存放（密码应密文存放，使用时解密），确保安全。
// 本示例以用户名和密码保存在环境变量中为例，运行本示例前请先在本地环境中设置环境变量（环境变量名称请根据自身情况进行设置）EXAMPLE_USERNAME_ENV和EXAMPLE_PASSWORD_ENV。
// $ip、$port、database需要用户自行修改。
// 逻辑复制类PGReplicationStream为非线程安全类，并发调用可能导致数据异常。
import com.huawei.gaussdb.jdbc.PGProperty;
import com.huawei.gaussdb.jdbc.jdbc.PgConnection;
import com.huawei.gaussdb.jdbc.replication.LogSequenceNumber;
import com.huawei.gaussdb.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.gaussdb.jdbc.Driver";
        //此处配置数据库IP以及端口，这里的端口为haPort（High Availability Port，高可用端口），haPort通常默认是所连接的DN端口号+1。
        String sourceURL = "jdbc:gaussdb://$ip:$port/database";
        
        //默认逻辑复制槽的名称是：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, 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")
                        .make();
            }else if(testMode == TEST_MODE_START_REPL) {
                //开启此模式前需要创建复制槽
                //waitLSN 表示日志序列号，用于确定逻辑复制开始的位置。此处的值仅为示例，请根据业务场景及数据库当前实际情况自行替换
                LogSequenceNumber waitLSN = LogSequenceNumber.valueOf("6F/E3C53568"); //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", 4) //解码线程并行度
                        .withSlotOption("white-table-list", "public.t1,public.t2") //白名单列表，根据实际需求配置是否开启白名单
                        // .withSlotOption("standby-connection", true) //强制备机解码
                        .withSlotOption("decode-style", "t") //解码格式
                        .withSlotOption("sending-batch", 0) //批量发送解码结果
                        .withSlotOption("max-txn-in-memory", 100) //单个解码事务落盘内存阈值为100MB
                        .withSlotOption("max-reorderbuffer-in-memory", 2) //正在处理的解码事务落盘内存阈值为2GB
                        
                        .withSlotOption("exclude-users", "userA") //不返回用户userA执行事务的逻辑日志
                        .withSlotOption("include-user", false) //事务BEGIN逻辑日志不携带用户名字
                        .withSlotOption("enable-heartbeat", true) // 开启心跳日志
                        .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，根据业务实际情况调用以下接口，每调用一次以下接口将触发数据库复制槽落盘逻辑，可能影响解码性能，建议每10s左右调用一次
                    //  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;
        } finally {
            try {
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
// 注意：如上代码不能直接读取二进制格式的逻辑日志，需按二进制格式编码规则读取。
```
text格式（即't'格式）解码结果示例如下：
```
BEGIN CSN: 2014 first_lsn: 0/2816A28
table public t1 INSERT: a[integer]:1 b[integer]:2 c[text]:'hello'
COMMIT XID: 15504
BEGIN CSN: 2015 first_lsn: 0/2816C20
table public t1 UPDATE: old-key: a[integer]:1 b[integer]:2 c[text]:'hello' new-tuple: a[integer]:1 b[integer]:5 c[text]:'hello'
COMMIT XID: 15505
BEGIN CSN: 2016 first_lsn: 0/2816D60
table public t1 DELETE: a[integer]:1 b[integer]:5 c[text]:'hello'
COMMIT XID: 15506
```
json格式（即'j'格式）解码结果示例如下：
```
BEGIN CSN: 2014 first_lsn: 0/2816A28
{"table_name":"public.t1","op_type":"INSERT","columns_name":["a","b","c"],"columns_type":["integer","integer","text"],"columns_val":["1","2","'hello'"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
COMMIT XID: 15504
BEGIN CSN: 2015 first_lsn: 0/2816C20
{"table_name":"public.t1","op_type":"UPDATE","columns_name":["a","b","c"],"columns_type":["integer","integer","text"],"columns_val":["1","5","'hello'"],"old_keys_name":["a","b","c"],"old_keys_type":["integer","integer","text"],"old_keys_val":["1","2","'hello'"]}
COMMIT XID: 15505
BEGIN CSN: 2016 first_lsn: 0/2816D60
{"table_name":"public.t1","op_type":"DELETE","columns_name":[],"columns_type":[],"columns_val":[],"old_keys_name":["a","b","c"],"old_keys_type":["integer","integer","text"],"old_keys_val":["1","5","'hello'"]}
COMMIT XID: 15506
```