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;