数据同步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;