。

这种限制导致数据推送过程受阻#xff0c;需要相应的调整或处理。

二、解决方法

在Canal推送数据变更信息至MQ消息队列时我们遇到了特定问题尤其是当消息体的大小超过了MQ所允许的最大限制。

这种限制导致数据推送过程受阻需要相应的调整或处理。

二、解决方法

采用Canal自定义客户端的方式将接收到的消息进行压缩处理并将消息转换为MQ消息格式。

经过处理的消息将被推送到原有的MQ主题与原有消息处理兼容从而在提升数据处理效率的同时确保既有的消费者逻辑得以保持不变实现平滑过渡与升级。

三、例子

dependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.client/artifactIdversion1.1.5/version

/dependency

dependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.protocol/artifactIdversion1.1.5/version

/dependency

cn.ewan.skyeye.log.audit.support.canal;import

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

import

java.util.zip.GZIPOutputStream;/***

Description:*

CanalConnectors.newSingleConnector(new

port),destination,

);connector.connect();connector.subscribe(FILTER_REGEX);}public

void

connector.getWithoutAck(batchSize);long

batchId

flatMessages){System.out.println(compress(flatMessage));}System.out.println(FlatMessage:

JSONUtil.toJsonStr(flatMessages));//

确认消息已被处理connector.ack(batchId);//

暂停一秒避免循环过快Thread.sleep(1000);}}public

static

EntryRowData[rawEntries.size()];int

0;for

CanalEntry.Entry.parseFrom(byteString);CanalEntry.RowChange

rowChange

CanalEntry.RowChange.parseFrom(entry.getStoreValue());datas[index]

new

EntryRowData();datas[index].entry

rowChange;}

(InvalidProtocolBufferException

{throw

EntryRowData[message.getEntries().size()];int

0;for

CanalEntry.RowChange.parseFrom(entry.getStoreValue());datas[index]

new

EntryRowData();datas[index].entry

rowChange;}

(InvalidProtocolBufferException

{throw

messageConverter(EntryRowData[]

datas,

entryRowData.entry;CanalEntry.RowChange

rowChange

CanalEntry.EntryType.TRANSACTIONBEGIN||

entry.getEntryType()

CanalEntry.EntryType.TRANSACTIONEND)

{continue;}//

flatMessageCanalEntry.EventType

eventType

rowChange.getEventType();FlatMessage

flatMessage

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

{MapString,

LinkedHashMap();ListCanalEntry.Column

columns;if

rowData.getBeforeColumnsList();}

else

rowData.getAfterColumnsList();}for

(CanalEntry.Column

{flatMessage.addPkName(column.getName());}sqlType.put(column.getName(),

column.getSqlType());mysqlType.put(column.getName(),

(column.getIsNull())

{updateSet.add(column.getName());}}hasInitPkNames

true;if

rowData.getBeforeColumnsList())

{if

(updateSet.contains(column.getName()))

{if

update操作将记录修改前的值old.add(rowOld);}}if

(!sqlType.isEmpty())

{flatMessage.setSqlType(sqlType);}if

{flatMessage.setMysqlType(mysqlType);}if

(!data.isEmpty())

{flatMessage.setOld(old);}}}return

static

printColumn(ListCanalEntry.Column

columns)

{System.out.println(column.getName()

String

JSONUtil.toJsonStr(flatMessage);//

new

ByteArrayOutputStream();GZIPOutputStream

gzipOutputStream

GZIPOutputStream(byteArrayOutputStream);try

{gzipOutputStream.write(jsonString.getBytes(UTF-8));}

finally

{gzipOutputStream.close();}byte[]

compressedData

byteArrayOutputStream.toByteArray();//

解压缩数据并还原为JSON字符串ByteArrayInputStream

byteArrayInputStream

ByteArrayInputStream(compressedData);GZIPInputStream

gzipInputStream

GZIPInputStream(byteArrayInputStream);ByteArrayOutputStream

new

{decompressedOutputStream.write(buffer,

len);}}

{gzipInputStream.close();}decompressedJsonString

decompressedOutputStream.toString(UTF-8);}

catch