Canal 解析与 Spring Boot 整合实战

一、Canal 简介

1.1 Canal 是什么？

Canal 是阿里巴巴开源的一款基于 MySQL 数据库增量日志解析（Binlog）中间件，它模拟 MySQL 的从机（Slave）行为，监听 MySQL 主机的二进制日志（Binlog），并解析出数据变更事件（DML 和 DDL），然后将这些事件转发给下游应用，从而实现数据的实时同步和处理。

1.2 Canal 的应用场景

数据同步：将 MySQL 数据库的变更数据实时同步到其他存储系统，如 Redis、Elasticsearch 等。
数据备份：实时备份 MySQL 数据库的变更数据，用于数据恢复或异地备份。
数据一致性：在分布式系统中，保证 MySQL 数据库和缓存（如 Redis）的数据一致性。
业务解耦：通过消息队列（如 Kafka）将数据变更事件传递给下游应用，实现业务系统的解耦。

二、Canal 工作原理

2.1 MySQL 主从复制原理

在介绍 Canal 的工作原理之前，我们先复习一下 MySQL 的主从复制原理。MySQL 主从复制主要涉及以下几个步骤：

主库记录二进制日志：主库将数据变更操作（如 INSERT、UPDATE、DELETE）记录到二进制日志（Binlog）中。
从库读取二进制日志：从库连接到主库，请求二进制日志，并将日志内容写入到本地的中继日志（Relay Log）中。
从库应用中继日志：从库的 SQL 线程读取中继日志，并将日志中的数据变更操作应用到从库的数据库中。

2.2 Canal 模拟从库行为

Canal 模拟了 MySQL 从库的行为，通过以下步骤实现数据的监听和解析：

连接到 MySQL 主库：Canal 以从库的身份连接到 MySQL 主库，并请求二进制日志。
解析二进制日志：Canal 解析二进制日志中的数据变更事件（DML 和 DDL），并将其转换为 Canal 自定义的事件格式。
转发数据变更事件：Canal 将解析后的数据变更事件转发给下游应用，如 Redis、Kafka 等。

2.3 Canal 的核心组件

Canal Server：负责与 MySQL 主库建立连接，监听二进制日志，并解析数据变更事件。
Canal Client：负责接收 Canal Server 转发的数据变更事件，并进行相应的处理。
Canal Filter：用于过滤二进制日志中的数据变更事件，只处理感兴趣的表和字段。

三、Canal 安装与配置

3.1 开启 MySQL 二进制日志

在使用 Canal 之前，需要确保 MySQL 的二进制日志已经开启。可以通过以下命令查看二进制日志是否开启：

SHOW VARIABLES LIKE 'log_bin';

如果 log_bin 的值为 OFF，则需要在 MySQL 配置文件中开启二进制日志：

[mysqld]
log-bin=mysql-bin

3.2 创建 Canal 用户

为了保证数据安全，建议为 Canal 创建一个专用的 MySQL 用户，并授予其必要的权限：

CREATE USER canal@'%' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT, SUPER ON *.* TO 'canal'@'%';
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
FLUSH PRIVILEGES;

3.3 使用 Docker 安装 Canal

Canal 提供了 Docker 镜像，可以通过以下命令安装并启动 Canal：

docker pull canal/canal-server:v1.1.5docker run -p 11111:11111 --name canal \
-e canal.destinations=tingshuTopic \
-e canal.instance.master.address=192.168.200.130:3306  \
-e canal.instance.dbUsername=canal  \
-e canal.instance.dbPassword=canal  \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false  \
-e canal.instance.filter.regex=.*\\..* \
-d canal/canal-server:v1.1.5

四、Spring Boot 整合 Canal

4.1 创建 Spring Boot 工程

使用 Spring Initializr 创建一个新的 Spring Boot 工程

4.2 添加依赖

在 pom.xml 文件中添加以下依赖：

目前canal不支持jdk17，变成jdk8版本

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.atguigu</groupId><artifactId>service-cdc</artifactId><version>1.0-SNAPSHOT</version><packaging>jar</packaging><name>service-cdc</name><url>http://maven.apache.org</url><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.3.6.RELEASE</version><relativePath/></parent><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><!--web 需要启动项目--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>top.javatool</groupId><artifactId>canal-spring-boot-starter</artifactId><version>1.2.1-RELEASE</version></dependency><!--    起到监听的作用     --><dependency><groupId>javax.persistence</groupId><artifactId>persistence-api</artifactId><version>1.0</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency></dependencies>
</project>

4.3 配置 Canal 和 Redis

一个canal服务器有很多个客户端，每个客户端有自己的通道名字

在 application-dev.yml 文件中配置 Canal 和 Redis 的连接信息：

server:port: 7080
#canal配置
canal:destination: tingshuTopic #Canal服务端发送数据的话题名称跟上面容器里参数destinations的一样server: 192.168.200.130:11111spring:redis:host: 192.168.200.130port: 6379

4.4 创建实体类

通过实体类监听到mysql变化的数据，但因为不同表的数据都不一样，所以每个实体类的字段都不一样，但是每个表都会有id，所以在实体类中加上变化字段的id

java">import lombok.Data;
import javax.persistence.Column;@Data
public class CDCEntity {// 注意Column 注解必须是persistence包下的,表示监听表中的一个字段@Column(name = "id")private Long id;
}

4.5 创建 Canal 处理类

创建一个类实现 EntryHandler 接口，用于处理 Canal 解析的数据变更事件：它定义了三个方法，分别对应 INSERT、UPDATE 和 DELETE 操作：

java">import com.alibaba.otter.canal.client.adapter.support.EntryHandler;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;@Slf4j
@Component
@CanalTable("album_info")
public class AlbumInfoCdcHandler implements EntryHandler<CDCEntity> {@Autowiredprivate RedisTemplate redisTemplate;@Overridepublic void insert(CDCEntity cdcEntity) {log.info("监听到数据添加，ID: {}", cdcEntity.getId());}@Overridepublic void update(CDCEntity before, CDCEntity after) {log.info("监听到数据更新，ID: {}", after.getId());String key = "album:info:" + after.getId();redisTemplate.delete(key);}@Overridepublic void delete(CDCEntity cdcEntity) {log.info("监听到数据删除，ID: {}", cdcEntity.getId());}
}

实例：

java">@Slf4j
@Component
@CanalTable("album_info") 监听变更表
public class AlbumInfoCdcHandler implements EntryHandler<CDCEntity> {@Autowiredprivate RedisTemplate redisTemplate;//mysql执行添加操作，这个方法执行public void insert(CDCEntity cdcEntity) {log.info("监听到数据修改,ID:{}", cdcEntity.getId());}//mysql执行修改操作，这个方法执行public void update(CDCEntity before, CDCEntity after) {log.info("监听到数据修改,ID:{}", after.getId());String key = "album:info:" + after.getId();redisTemplate.delete(key);}//mysql执行删除操作，这个方法执行public void delete(CDCEntity cdcEntity) {log.info("监听到数据修改,ID:{}", cdcEntity.getId());}
}

五、总结

Canal 是一款强大的 MySQL 数据库增量日志解析中间件，通过模拟 MySQL 从库的行为，实现数据的实时同步和处理。在本文中，我们详细介绍了 Canal 的工作原理、安装配置方法以及如何与 Spring Boot 进行整合。通过 Canal，我们可以轻松地实现 MySQL 数据库与 Redis 等其他存储系统的数据一致性，为分布式系统的开发提供了有力的支持。

希望本文对你有所帮助。如果有任何问题或建议，欢迎在评论区留言。