【canal】canal同步msyql到redis

java层面 canal client配置pom文件

<dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> <version>2.13.2</version> </dependency> <dependency> <groupId>com.google.guava</groupId> <artifactId>guava</artifactId> <version>33.5.0-jre</version> </dependency> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.7</version> <!-- 使用最新版本 --> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.47</version> </dependency> <!-- Redisson --> <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.23.5</version> </dependency> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.protocol</artifactId> <version>1.1.7</version> </dependency>

java client

1.将接收到的canal数据2进制转为10进制数据

2.redis client需要2进制转10进制，否则key是10进制，内部还是2进制

package org.example.cannal.redis; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.CanalEntry.*; import org.redisson.api.RMap; import org.redisson.api.RedissonClient; import org.redisson.config.Config; import java.net.InetSocketAddress; import java.util.HashMap; import java.util.List; import java.util.Map; public class CanalBinaryParseWithRedisson { public static void main(String[] args) { // 1. 创建 Canal 连接 CanalConnector connector = CanalConnectors.newSingleConnector( new InetSocketAddress("127.0.0.1", 11111), "example", "", "" ); // 2. 创建 Redisson 客户端 RedissonClient redisson = createRedissonClient(); try { connector.connect(); connector.subscribe(".*\\..*"); connector.rollback(); while (true) { Message message = connector.getWithoutAck(100); // 每次取100条 long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { try { Thread.sleep(1000); } catch (InterruptedException ignored) {} } else { handleEntries(message.getEntries(), redisson); connector.ack(batchId); // 确认消费 } } } finally { connector.disconnect(); redisson.shutdown(); } } private static void handleEntries(List<Entry> entrys, RedissonClient redisson) { for (Entry entry : entrys) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { continue; } RowChange rowChange; try { // 关键：把二进制解析成 RowChange rowChange = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("解析 RowChange 失败", e); } EventType eventType = rowChange.getEventType(); String tableName = entry.getHeader().getTableName(); for (RowData rowData : rowChange.getRowDatasList()) { if (eventType == EventType.INSERT || eventType == EventType.UPDATE) { Map<String, String> dataMap = new HashMap<>(); String id = null; for (Column column : rowData.getAfterColumnsList()) { dataMap.put(column.getName(), column.getValue()); if ("id".equals(column.getName())) { id = column.getValue(); } } // 用 Redisson 写入 Redis Hash RMap<String, String> map = redisson.getMap(tableName + ":" + id); map.putAll(dataMap); System.out.println("写入 Redis: " + tableName + ":" + id + " -> " + dataMap); } else if (eventType == EventType.DELETE) { String id = null; for (Column column : rowData.getBeforeColumnsList()) { if ("id".equals(column.getName())) { id = column.getValue(); break; } } // 删除 Redis Key redisson.getMap(tableName + ":" + id).delete(); System.out.println("删除 Redis: " + tableName + ":" + id); } } } } private static RedissonClient createRedissonClient() { Config config = new Config(); // 1. 配置Redis连接（单节点为例） config.useSingleServer() .setAddress("redis://127.0.0.1:6379") // .setPassword("你的密码") // 无密码则省略 .setDatabase(1); // 2. 关键：设置序列化器为StringCodec（明文字符串序列化） config.setCodec(new org.redisson.client.codec.StringCodec()); return org.redisson.Redisson.create(config); } }