Canal客户端开发
创建client_demo项目
Maven依赖
com.alibaba.otter
canal.client
1.0.24
com.alibaba
fastjson
1.2.58
- 创建Connector
- 连接Cannal服务器，并订阅
- 解析Canal消息，并打印
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
sqlType [jdbc type]
name [column name]
isKey [是否为主键]
updated [是否发生过变更]
isNull [值是否为null]
value [具体的内容，注意为string文本]
参考代码：
public class CanalClientEntrance {
public static void main(String[] args) {
// 1. 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.88.120",
11111), "example", "canal", "canal");
// 指定一次性获取数据的条数
int batchSize = 5 * 1024;
boolean running = true;
try {
while(running) {
// 2. 建立连接
connector.connect();
// 回滚上次的get请求，重新获取数据
connector.rollback();
// 订阅匹配日志
connector.subscribe("itcast_shop.*");
while(running) {
// 批量拉取binlog日志，一次性获取多条数据
Message message = connector.getWithoutAck(batchSize);
// 获取batchId
long batchId = message.getId();
// 获取binlog数据的条数
int size = message.getEntries().size();
if(batchId == -1 || size == 0) {
}
else {
printSummary(message);
}
// 确认指定的batchId已经消费成功
connector.ack(batchId);
}
}
} finally {
// 断开连接
connector.disconnect();
}
}
private static void printSummary(Message message) {
// 遍历整个batch中的每个binlog实体
for (CanalEntry.Entry entry : message.getEntries()) {
// 事务开始
if(entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
// 获取binlog文件名
String logfileName = entry.getHeader().getLogfileName();
// 获取logfile的偏移量
long logfileOffset = entry.getHeader().getLogfileOffset();
// 获取sql语句执行时间戳
long executeTime = entry.getHeader().getExecuteTime();
// 获取数据库名
String schemaName = entry.getHeader().getSchemaName();
// 获取表名
String tableName = entry.getHeader().getTableName();
// 获取事件类型 insert/update/delete
String eventTypeName = entry.getHeader().getEventType().toString().toLowerCase();
System.out.println("logfileName" + ":" + logfileName);
System.out.println("logfileOffset" + ":" + logfileOffset);
System.out.println("executeTime" + ":" + executeTime);
System.out.println("schemaName" + ":" + schemaName);
System.out.println("tableName" + ":" + tableName);
System.out.println("eventTypeName" + ":" + eventTypeName);
CanalEntry.RowChange rowChange = null;
try {
// 获取存储数据，并将二进制字节数据解析为RowChange实体
rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (InvalidProtocolBufferException e) {
e.printStackTrace();
}
// 迭代每一条变更数据
for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
// 判断是否为删除事件
if(entry.getHeader().getEventType() == CanalEntry.EventType.DELETE) {
System.out.println("---delete---");
printColumnList(rowData.getBeforeColumnsList());
System.out.println("---");
}
// 判断是否为更新事件
else if(entry.getHeader().getEventType() == CanalEntry.EventType.UPDATE) {
System.out.println("---update---");
printColumnList(rowData.getBeforeColumnsList());
System.out.println("---");
printColumnList(rowData.getAfterColumnsList());
}
// 判断是否为插入事件
else if(entry.getHeader().getEventType() == CanalEntry.EventType.INSERT) {
System.out.println("---insert---");
printColumnList(rowData.getAfterColumnsList());
System.out.println("---");
}
}
}
}
// 打印所有列名和列值
private static void printColumnList(List columnList) {
for (CanalEntry.Column column : columnList) {
System.out.println(column.getName() + "\t" + column.getValue());
}
}
}
- 复制上述代码，将binlog日志封装在一个Map结构中，使用fastjson转换为JSON格式
参考代码：
// binlog解析为json字符串
private static String binlogToJson(Message message) throws InvalidProtocolBufferException {
// 1. 创建Map结构保存最终解析的数据
Map rowDataMap = new HashMap();
// 2. 遍历message中的所有binlog实体
for (CanalEntry.Entry entry : message.getEntries()) {
// 只处理事务型binlog
if(entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
// 获取binlog文件名
String logfileName = entry.getHeader().getLogfileName();
// 获取logfile的偏移量
long logfileOffset = entry.getHeader().getLogfileOffset();
// 获取sql语句执行时间戳
long executeTime = entry.getHeader().getExecuteTime();
// 获取数据库名
String schemaName = entry.getHeader().getSchemaName();
// 获取表名
String tableName = entry.getHeader().getTableName();
// 获取事件类型 insert/update/delete
String eventType = entry.getHeader().getEventType().toString().toLowerCase();
rowDataMap.put("logfileName", logfileName);
rowDataMap.put("logfileOffset", logfileOffset);
rowDataMap.put("executeTime", executeTime);
rowDataMap.put("schemaName", schemaName);
rowDataMap.put("tableName", tableName);
rowDataMap.put("eventType", eventType);
// 封装列数据
Map columnDataMap = new HashMap();
// 获取所有行上的变更
CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
List columnDataList = rowChange.getRowDatasList();
for (CanalEntry.RowData rowData : columnDataList) {
if(eventType.equals("insert") || eventType.equals("update")) {
for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
columnDataMap.put(column.getName(), column.getValue());
}
}
else if(eventType.equals("delete")) {
for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
columnDataMap.put(column.getName(), column.getValue());
}
}
}
rowDataMap.put("columns", columnDataMap);
}
return JSON.toJSONString(rowDataMap);
}
