一、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-bin3.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.5
docker 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:11111
spring:
redis:
host: 192.168.200.130
port: 63794.4 创建实体类
通过实体类监听到mysql变化的数据，但因为不同表的数据都不一样，所以每个实体类的字段都不一样，但是每个表都会有id，所以在实体类中加上变化字段的id
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 操作：
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> {
@Autowired
private RedisTemplate redisTemplate;
@Override
public void insert(CDCEntity cdcEntity) {
log.info("监听到数据添加，ID: {}", cdcEntity.getId());
}
@Override
public void update(CDCEntity before, CDCEntity after) {
log.info("监听到数据更新，ID: {}", after.getId());
String key = "album:info:" + after.getId();
redisTemplate.delete(key);
}
@Override
public void delete(CDCEntity cdcEntity) {
log.info("监听到数据删除，ID: {}", cdcEntity.getId());
}
}实例：
@Slf4j
@Component
@CanalTable("album_info") 监听变更表
public class AlbumInfoCdcHandler implements EntryHandler<CDCEntity> {
@Autowired
private 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 等其他存储系统的数据一致性，为分布式系统的开发提供了有力的支持。
希望本文对你有所帮助。如果有任何问题或建议，欢迎在评论区留言。
![LiteratureReading:[2016] Enriching Word Vectors with Subword Information](https://i-blog.csdnimg.cn/direct/9d68ad484331416eb9c2b5f6805ab92c.png)
