用spring boot 如何实现 mysql canal数据同步到redis的demo?

要使用 Spring Boot 和 Canal 实现 MySQL 数据同步到 Redis，通常的流程是通过 Canal 监听 MySQL 的 binlog 日志，然后将数据库中的变更（增、删、改）同步到 Redis 中。 ### 实现步骤： 1. **准备工作**： - 安装并配置 Canal Server，用于监听 MySQL 数据库的 binlog 日志。 - 创建 Spring Boot 项目，集成 Canal 客户端，监听 Canal Server，并将数据库变化的数据同步到 Redis。 ### 1. **MySQL 和 Canal 配置** - **MySQL**：需要开启 binlog 日志支持，并设置 binlog 格式为 `ROW` 模式。 在 `my.cnf` 或 `my.ini` 配置文件中添加： ```ini [mysqld] log-bin=mysql-bin binlog-format=ROW server-id=1 ``` - **Canal Server**： - 下载并安装 Canal Server，配置 MySQL 数据库的访问权限，确保 Canal Server 能够读取 MySQL 的 binlog 日志。 - 配置 Canal Server 的 `instance.properties` 文件，指定 MySQL 数据库的连接信息。 ### 2. **创建 Spring Boot 项目** 创建一个简单的 Spring Boot 项目，并引入相关的依赖： ```xml <dependencies> <!-- Canal 客户端 --> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.5</version> </dependency> <!-- Redis 依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <!-- Spring Boot Web 依赖 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> </dependencies> ``` ### 3. **配置 Canal 客户端** 首先，我们需要编写 Canal 客户端的配置，让它能够连接到 Canal Server，并监听 MySQL 的 binlog 日志。 创建 Canal 配置类： ```java 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.Message; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.net.InetSocketAddress; @Configuration public class CanalClientConfig { @Value("${canal.server.hostname}") private String canalHostName; @Value("${canal.server.port}") private int canalPort; @Value("${canal.destination}") private String destination; @Value("${canal.username}") private String username; @Value("${canal.password}") private String password; @Bean public CanalConnector canalConnector() { // 创建与 Canal Server 的连接 return CanalConnectors.newSingleConnector( new InetSocketAddress(canalHostName, canalPort), destination, username, password); } } ``` ### 4. **监听 MySQL 变更** 创建 Canal 客户端监听器，它负责从 Canal Server 中获取 binlog 数据，并同步到 Redis 中。 ```java import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; import javax.annotation.PostConstruct; import java.util.List; @Service public class CanalService { private static final Logger logger = LoggerFactory.getLogger(CanalService.class); @Autowired private CanalConnector canalConnector; @Autowired private StringRedisTemplate redisTemplate; @PostConstruct public void start() { new Thread(() -> { try { canalConnector.connect(); canalConnector.subscribe(".*\\..*"); // 订阅所有数据库的所有表 canalConnector.rollback(); while (true) { // 获取指定数量的数据 Message message = canalConnector.getWithoutAck(100); long batchId = message.getId(); int size = message.getEntries().size(); if (batchId != -1 && size != 0) { processEntries(message.getEntries()); canalConnector.ack(batchId); // 提交确认 } else { Thread.sleep(1000); // 没有数据时休眠 1 秒 } } } catch (Exception e) { logger.error("Canal 数据同步异常", e); } finally { canalConnector.disconnect(); } }).start(); } private void processEntries(List<CanalEntry.Entry> entries) { for (CanalEntry.Entry entry : entries) { if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) { try { CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); CanalEntry.EventType eventType = rowChange.getEventType(); // 处理每一行数据的变化 for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { if (eventType == CanalEntry.EventType.INSERT) { handleInsert(rowData); } else if (eventType == CanalEntry.EventType.DELETE) { handleDelete(rowData); } else if (eventType == CanalEntry.EventType.UPDATE) { handleUpdate(rowData); } } } catch (Exception e) { logger.error("解析 binlog 异常", e); } } } } private void handleInsert(CanalEntry.RowData rowData) { // 将插入的数据同步到 Redis String key = getKey(rowData.getAfterColumnsList()); String value = getValue(rowData.getAfterColumnsList()); redisTemplate.opsForValue().set(key, value); logger.info("插入数据: key={}, value={}", key, value); } private void handleDelete(CanalEntry.RowData rowData) { // 将删除的数据从 Redis 中删除 String key = getKey(rowData.getBeforeColumnsList()); redisTemplate.delete(key); logger.info("删除数据: key={}", key); } private void handleUpdate(CanalEntry.RowData rowData) { // 更新 Redis 中的数据 String key = getKey(rowData.getAfterColumnsList()); String value = getValue(rowData.getAfterColumnsList()); redisTemplate.opsForValue().set(key, value); logger.info("更新数据: key={}, value={}", key, value); } private String getKey(List<CanalEntry.Column> columns) { // 假设第一个字段是主键 return columns.get(0).getValue(); } private String getValue(List<CanalEntry.Column> columns) { // 简单拼接字段作为值 StringBuilder builder = new StringBuilder(); for (CanalEntry.Column column : columns) { builder.append(column.getName()).append("=").append(column.getValue()).append(" "); } return builder.toString().trim(); } } ``` ### 5. **Redis 配置** 在 `application.yml` 中配置 Redis 和 Canal Server 的信息： ```yaml spring: redis: host: localhost port: 6379 canal: server: hostname: localhost port: 11111 destination: example username: canal password: canal ``` ### 6. **启动 Canal Server** 确保 MySQL 和 Canal Server 已经配置并运行。启动 Spring Boot 应用后，Canal 会监听 MySQL binlog，当数据库发生变化时（如插入、删除、更新），对应的数据会同步到 Redis。 ### 总结 1. 通过 Canal，监听 MySQL 的 binlog 日志，捕获数据的变动。 2. 使用 Spring Boot 集成 Canal 客户端，将变动的数据实时同步到 Redis。 3. 该 Demo 中，示例代码实现了插入、删除和更新操作的同步，你可以根据需要进一步优化数据同步的逻辑。 这样你就实现了一个简单的 MySQL 数据同步到 Redis 的方案。