如何给一个网站做定时的更新,南昌市科协网站,网络广告策划内容,wordpress论坛用户一、背景
在Canal推送数据变更信息至MQ#xff08;消息队列#xff09;时#xff0c;我们遇到了特定问题#xff0c;尤其是当消息体的大小超过了MQ所允许的最大限制。这种限制导致数据推送过程受阻#xff0c;需要相应的调整或处理。
二、解决方法
采用Canal自定义客户…一、背景
在Canal推送数据变更信息至MQ消息队列时我们遇到了特定问题尤其是当消息体的大小超过了MQ所允许的最大限制。这种限制导致数据推送过程受阻需要相应的调整或处理。
二、解决方法
采用Canal自定义客户端的方式将接收到的消息进行压缩处理并将消息转换为MQ消息格式。经过处理的消息将被推送到原有的MQ主题与原有消息处理兼容从而在提升数据处理效率的同时确保既有的消费者逻辑得以保持不变实现平滑过渡与升级。
三、例子
!-- canal 客户端的集成 --
dependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.client/artifactIdversion1.1.5/version
/dependency
!-- 客户端通信 消息传递 --
dependencygroupIdcom.alibaba.otter/groupIdartifactIdcanal.protocol/artifactIdversion1.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;}ListFlatMessage 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()) {ListByteString 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 ListFlatMessage messageConverter(EntryRowData[] datas, long id) {ListFlatMessage 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()) {MapString, Integer sqlType new LinkedHashMap();MapString, String mysqlType new LinkedHashMap();ListMapString, String data new ArrayList();ListMapString, String old new ArrayList();SetString 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;}MapString, String row new LinkedHashMap();ListCanalEntry.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) {MapString, 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(ListCanalEntry.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();}
}
