Canal自定义客户端

一、背景

在Canal推送数据变更信息至MQ（消息队列）时，我们遇到了特定问题，尤其是当消息体的大小超过了MQ所允许的最大限制。这种限制导致数据推送过程受阻，需要相应的调整或处理。

二、解决方法

采用Canal自定义客户端的方式，将接收到的消息进行压缩处理，并将消息转换为MQ消息格式。经过处理的消息将被推送到原有的MQ主题，与原有消息处理兼容，从而在提升数据处理效率的同时，确保既有的消费者逻辑得以保持不变，实现平滑过渡与升级。

三、例子

<!-- canal 客户端的集成 -->
<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.5</version>
</dependency>
<!-- 客户端通信 消息传递 -->
<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.protocol</artifactId><version>1.1.5</version>
</dependency>

package cn.ewan.skyeye.log.audit.support.canal;import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;/*** @Description:* @Author: wulm* @Date: 2024/6/14 10:11* @Version:1.0*/
public class CustomCanalClient {private final CanalConnector connector;private final String FILTER_REGEX = "test_db.(test_table)";public CustomCanalClient(String destination, String host, int port) {connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port),destination, "", "");connector.connect();connector.subscribe(FILTER_REGEX);}public void listen() throws InterruptedException {while (true) {// 一次获取的变更事件数量int batchSize = 1000;Message message = connector.getWithoutAck(batchSize);long batchId = message.getId();EntryRowData[] entryRowData = buildMessageData(message);if (entryRowData == null) {continue;}List<FlatMessage> flatMessages = messageConverter(entryRowData, batchId);for (FlatMessage flatMessage : flatMessages){System.out.println(compress(flatMessage));}System.out.println("============FlatMessage: " + JSONUtil.toJsonStr(flatMessages));// 确认消息已被处理connector.ack(batchId);// 暂停一秒，避免循环过快Thread.sleep(1000);}}public static class EntryRowData {public CanalEntry.Entry entry;public CanalEntry.RowChange rowChange;}public static EntryRowData[] buildMessageData(Message message) {if (message.isRaw()) {List<ByteString> rawEntries = message.getRawEntries();final EntryRowData[] datas = new EntryRowData[rawEntries.size()];int i = 0;for (ByteString byteString : rawEntries) {final int index = i;try {CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString);CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());datas[index] = new EntryRowData();datas[index].entry = entry;datas[index].rowChange = rowChange;} catch (InvalidProtocolBufferException e) {throw new RuntimeException(e);}i++;}return datas;} else {final EntryRowData[] datas = new EntryRowData[message.getEntries().size()];int i = 0;for (CanalEntry.Entry entry : message.getEntries()) {final int index = i;try {CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());datas[index] = new EntryRowData();datas[index].entry = entry;datas[index].rowChange = rowChange;} catch (InvalidProtocolBufferException e) {throw new RuntimeException(e);}i++;}return datas;}}public static List<FlatMessage> messageConverter(EntryRowData[] datas, long id) {List<FlatMessage> flatMessages = new ArrayList<>();for (EntryRowData entryRowData : datas) {CanalEntry.Entry entry = entryRowData.entry;CanalEntry.RowChange rowChange = entryRowData.rowChange;// 如果有分区路由,则忽略begin/end事件if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN|| entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {continue;}// build flatMessageCanalEntry.EventType eventType = rowChange.getEventType();FlatMessage flatMessage = new FlatMessage(id);flatMessages.add(flatMessage);flatMessage.setDatabase(entry.getHeader().getSchemaName());flatMessage.setTable(entry.getHeader().getTableName());flatMessage.setIsDdl(rowChange.getIsDdl());flatMessage.setType(eventType.toString());flatMessage.setEs(entry.getHeader().getExecuteTime());flatMessage.setTs(System.currentTimeMillis());flatMessage.setSql(rowChange.getSql());
//            flatMessage.setGtid(entry.getHeader().getGtid());if (!rowChange.getIsDdl()) {Map<String, Integer> sqlType = new LinkedHashMap<>();Map<String, String> mysqlType = new LinkedHashMap<>();List<Map<String, String>> data = new ArrayList<>();List<Map<String, String>> old = new ArrayList<>();Set<String> updateSet = new HashSet<>();boolean hasInitPkNames = false;for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE&& eventType != CanalEntry.EventType.DELETE) {continue;}Map<String, String> row = new LinkedHashMap<>();List<CanalEntry.Column> columns;if (eventType == CanalEntry.EventType.DELETE) {columns = rowData.getBeforeColumnsList();} else {columns = rowData.getAfterColumnsList();}for (CanalEntry.Column column : columns) {if (!hasInitPkNames && column.getIsKey()) {flatMessage.addPkName(column.getName());}sqlType.put(column.getName(), column.getSqlType());mysqlType.put(column.getName(), column.getMysqlType());if (column.getIsNull()) {row.put(column.getName(), null);} else {row.put(column.getName(), column.getValue());}// 获取update为true的字段if (column.getUpdated()) {updateSet.add(column.getName());}}hasInitPkNames = true;if (!row.isEmpty()) {data.add(row);}if (eventType == CanalEntry.EventType.UPDATE) {Map<String, String> rowOld = new LinkedHashMap<>();for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {if (updateSet.contains(column.getName())) {if (column.getIsNull()) {rowOld.put(column.getName(), null);} else {rowOld.put(column.getName(), column.getValue());}}}// update操作将记录修改前的值old.add(rowOld);}}if (!sqlType.isEmpty()) {flatMessage.setSqlType(sqlType);}if (!mysqlType.isEmpty()) {flatMessage.setMysqlType(mysqlType);}if (!data.isEmpty()) {flatMessage.setData(data);}if (!old.isEmpty()) {flatMessage.setOld(old);}}}return flatMessages;}private static void printColumn(List<CanalEntry.Column> columns) {for (CanalEntry.Column column : columns) {System.out.println(column.getName() + " : " + column.getValue());}}private String compress(FlatMessage flatMessage) {String decompressedJsonString = null;try {String jsonString = JSONUtil.toJsonStr(flatMessage);// 压缩JSON字符串ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();GZIPOutputStream gzipOutputStream = new GZIPOutputStream(byteArrayOutputStream);try {gzipOutputStream.write(jsonString.getBytes("UTF-8"));} finally {gzipOutputStream.close();}byte[] compressedData = byteArrayOutputStream.toByteArray();// 解压缩数据并还原为JSON字符串ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(compressedData);GZIPInputStream gzipInputStream = new GZIPInputStream(byteArrayInputStream);ByteArrayOutputStream decompressedOutputStream = new ByteArrayOutputStream();try {byte[] buffer = new byte[1024];int len;while ((len = gzipInputStream.read(buffer)) != -1) {decompressedOutputStream.write(buffer, 0, len);}} finally {gzipInputStream.close();}decompressedJsonString = decompressedOutputStream.toString("UTF-8");} catch (Exception e) {e.printStackTrace();}return decompressedJsonString;}public static void main(String[] args) throws InterruptedException {CustomCanalClient client = new CustomCanalClient("example", "127.0.0.1", 11111);client.listen();}
}