數據同步canal客戶端

1、增量訂閱、消費設計

get/ack/rollback協議介紹：

① Message getWithoutAck(int batchSize)，允許指定batchSize，一次可以獲取多條，每次返回的對象為Message，包含的內容為：

　　a. batch id 唯一標識
　　b. entries 具體的數據對象，對應的數據對象格式：EntryProtocol.proto

② void rollback(long batchId)，顧命思議，回滾上次的get請求，重新獲取數據。基於get獲取的batchId進行提交，避免誤操作

③ void ack(long batchId)，顧命思議，確認已經消費成功，通知server刪除數據。基於get獲取的batchId進行提交，避免誤操作

2、數據對象格式：EntryProtocol.proto

Entry
    Header
        logfileName [binlog文件名]
        logfileOffset [binlog position]
        executeTime [binlog里記錄變更發生的時間戳]
        schemaName [數據庫實例]
        tableName [表名]
        eventType [insert/update/delete類型]
    entryType     [事務頭BEGIN/事務尾END/數據ROWDATA]
    storeValue     [byte數據,可展開，對應的類型為RowChange]
RowChange
isDdl        [是否是ddl變更操作，比如create table/drop table]
sql        [具體的ddl sql]
rowDatas    [具體insert/update/delete的變更數據，可為多條，1個binlog event事件可對應多條變更，比如批處理]
beforeColumns [Column類型的數組]
afterColumns [Column類型的數組]


Column
index        [column序號]
sqlType        [jdbc type]
name        [column name]
isKey        [是否為主鍵]
updated        [是否發生過變更]
isNull        [值是否為null]
value        [具體的內容，注意為文本]

insert只有after columns, delete只有before columns，而update則會有before / after columns數據.

3、client使用例子

3.1 創建Connector

a. 創建SimpleCanalConnector (直連ip，不支持server/client的failover機制)

CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),11111), destination, "", "");

b. 創建ClusterCanalConnector (基於zookeeper獲取canal server ip，支持server/client的failover機制）

CanalConnector connector = CanalConnectors.newClusterConnector("10.20.144.51:2181", destination, "", "");

c. 創建ClusterCanalConnector (基於固定canal server的地址，支持固定的server ip的failover機制，不支持client的failover機制

CanalConnector connector = CanalConnectors.newClusterConnector(Arrays.asList(new InetSocketAddress(AddressUtils.getHostIp(),11111)), destination,"", "");

如上可見，創建client connector的時候需要指定destination，即對應於一個instance，一個數據庫。所以canal client和數據庫是一一對應的關系。

3.2 get/ack/rollback使用

// 創建鏈接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canal_ip, 11111), destination, canal_username, canal_password);

try {

    // 連接canal，獲取數據
    connector.connect();
    connector.subscribe();
    connector.rollback();
    log.info("數據同步工程啟動成功，開始獲取數據");
    while (true) {
        
        // 獲取指定數量的數據
        Message message = connector.getWithoutAck(1000);

        // 數據批號
        long batchId = message.getId();

        // 獲取該批次數據的數量
        int size = message.getEntries().size();

        // 無數據
        if (batchId == -1 || size == 0) {

            // 等待1秒后重新獲取
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                log.error(e);
                Thread.currentThread().interrupt();
            }

            // 提交確認
            connector.ack(batchId);

            // 數據存在，執行方法
        } else {
            try {

　　　　　　　　　 // 處理數據
                HandleData.handleEntry(message.getEntries());

                // 提交確認
                connector.ack(batchId);
            } catch (KafkaException e) {
                log.error(e);

                // 處理失敗, 回滾數據
                connector.rollback(batchId);
            } catch (Exception e1) {
                log.error(e1);

                // 提交確認
                connector.ack(batchId);
            }
        }
    }
} catch (Exception e) {

    log.error(e);
} finally {

    // 斷開連接
    connector.disconnect();
}

 處理數據的方法封裝到HandleData類中，且看handleEntry如何處理

// 獲取日志行
RowChange rowChage = null;
try {
    rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
    log.error(e);
}

// 獲取執行事件類型
EventType eventType = rowChage.getEventType();

// 日志打印，數據明細
log.info(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry
        .getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader()
        .getSchemaName(), entry.getHeader().getTableName(), eventType));

　// 獲取表名
　String tableName = entry.getHeader().getTableName();


// 遍歷日志行，執行任務
for (RowData rowData : rowChage.getRowDatasList()) {
    Map<String, Object> data;

    // 刪除操作
    if (eventType == EventType.DELETE) {

        // 解析數據
        data = DataUtils.parseData(tableName, "delete", rowData.getBeforeColumnsList());

        // 插入操作
    } else if (eventType == EventType.INSERT) {

        // 解析數據
        data = DataUtils.parseData(tableName, "insert", rowData.getAfterColumnsList());

        // 更新操作
    } else {

        // 解析數據
        data = DataUtils.parseData(tableName, "update", rowData.getAfterColumnsList());
    }

    // 數據解析成功
    if (data != null && data.size() > 0) {


        // 內容轉接json格式發送
        JSONObject json = JSONObject.fromObject(data);
        try {
            Productor.send("canal_" + tableName = "_topic", json.toString(), tableName + "|" + data.get("canal_kafka_key"));
        } catch (Exception e) {
            throw new KafkaException("kafka發送異常：" + e);
        }

        log.info("數據成功發送kafka");
    }
}

Entry數據被解析成Map格式數據，然后轉為json字符串，發到kafka。為什么要借用消息中間件kafka呢，不用kafka可以嗎？當然可以，直接寫數據同步的邏輯沒有問題。但是如果一個數據用到多個業務場景，勢必導致一個類中有多套同步邏輯，對於后期的維護很不利，多套業務摻雜在一起勢必會互相影響。合理的做法應該是業務隔離，每套業務都能接受到數據變更的消息，然后做自己需要的同步，這樣就需要在數據接受和數據處理形成1對n的關系。消息中間件的消息接受和消費模型正好可以完成這個功能。

一個canal client的消息分發給多個kafka消費者消費。每個kafka消費者代表一種業務場景，架構清晰、利於維護，同時一個kafka消費者可以消費多個canal client的topic。

上面的解析數據邏輯比較簡單，將list解析成map

Map<String, Object> result = new HashMap<String, Object>();
try {
    int index = 0;
    for (Column column : columns) {
        String value = column.getIsNull() ? null : column.getValue();

        // kafka在消息為10K時吞吐量達到最大
        if (value != null && value.length() > 10240) {
            value = value.substring(0, 10240);
        }
        if (index == 0) {
            result.put("canal_kafka_key", value);
        }
        result.put(column.getName(), value);
        index++;
    }
    result.put("operate_type", "delete"||"insert"||"update");
} catch (Exception e) {
    log.error(e);
}
if (logStr.lastIndexOf(",") == logStr.length() - 1) {
    logStr = logStr.substring(0, logStr.length() - 1);
}
return result;