准备工作

什么是 Canal

Canal 是阿里巴巴开源的基于数据库增量日志解析的数据同步和订阅组件，支持 MySQL、Oracle 等主流数据库。
它通过解析数据库的增量日志，将数据库的变更事件转发给下游消费者。

Canal 在 Spring Boot 中的作用和优势

Spring Boot 是一种用于简化 Spring 应用开发的框架，提供了快速、方便的构建 Spring 应用的方式。
Canal 在 Spring Boot 中可以作为一个可靠的、高性能的监听工具，实时捕获数据库的变更事件。
通过集成 Canal，可以在 Spring Boot 中方便地实现对 MySQL 表数据的监听，实现细粒度的数据同步、订阅等功能。

准备工作

安装和配置 MySQL 数据库

安装 MySQL 数据库，并确保数据库的正常运行。
开启binlog日志、配置需要监听的数据库；编辑 MySQL 的配置文件 my.cnf 或 my.ini，在 [mysqld]段中添加如下配置：

# 日志文件名
log-bin=mysql-bin
# 配置日志记录模式为“行级模式”
binlog_format=ROW
# 需要监听的数据库1
binlog-do-db=ddz
# 需要监听的数据库2 
binlog-do-db=ddz1

重启数据库，验证是否开启成功。

# 查看是否开启日志
show variables like '%log_bin%';

在这里插入图片描述

# 查看日志记录模式
show variables like '%binlog_format%';

在这里插入图片描述

安装Canal

在 Canal官网下载服务项目。

以下是对 Canal 的几个组件的详细说明：

canal.adapter 是 Canal 的一个子项目，主要用于数据同步和转换的适配器。
canal.admin 是 Canal 的后台管理系统。
canal.deployer 提供了一键部署和管理的功能，可以简化 Canal 的部署和运维工作。
canal.example 是 Canal 的示例项目。

解压文件，修改\canal.deployer-1.1.7-SNAPSHOT\conf\canal.properties文件；选择服务模式，我们这里需要在项目中使用则选择tcp。

# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp

修改\canal.deployer-1.1.7-SNAPSHOT\conf\example\instance.properties文件。

# 需要监听的数据库连接地址
canal.instance.master.address=127.0.0.1:3306
# 用户名
canal.instance.dbUsername=数据库用户名
# 密码
canal.instance.dbPassword=数据库密码
# 需要监听的数据库正则，当前是监听所有的数据库
canal.instance.filter.regex=.*\\..*
# 不监听的数据库正则
canal.instance.filter.black.regex=mysql\\.slave_.*

启动Canal项目，进入\canal.deployer-1.1.7-SNAPSHOT\bin，启动 startup.bat 或 startup.sh。

弹出下面页面说明启动成功：

项目集成

导入依赖

我们这里导入的是大佬封装好的依赖，我们直接使用就行。

<dependency><groupId>top.javatool</groupId><artifactId>canal-spring-boot-starter</artifactId><version>1.2.1-RELEASE</version>
</dependency>

添加配置信息

canal:server: 127.0.0.1:11111 # 本地启动的canal服务，默认启动的监听端口是11111destination: example # 配置好的监听模块

监听模块的名字和目录名一致，在canal配置文件中也配置的相同。
在这里插入图片描述

创建监听类

注解的使用：@CanalTable(value = “数据库表名”)

import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;
@Slf4j
@Component
@CanalTable(value = "user")
public class UserHandler implements EntryHandler<User> {@Overridepublic void insert(User user) {log.info("User_insert：{}", user.toString());}@Overridepublic void update(User before, User after) {log.info("User_update_before：{}", before.toString());log.info("User_update_after：{}", after.toString());}@Overridepublic void delete(User user) {log.info("User_delete：{}", user.toString());}
}