SpringBoot 集成 Canal 实现监听MySQL表数据

准备工作

什么是 Canal

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

Canal 在 Spring Boot 中的作用和优势

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

准备工作

安装和配置 MySQL 数据库

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

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

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

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

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

安装Canal

1. 在 Canal官网 下载服务项目。
   

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

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

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

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

3. 修改\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_.*
1
2
3
4
5
6
7
8
9
10

4. 启动Canal项目，进入\canal.deployer-1.1.7-SNAPSHOT\bin，启动 startup.bat 或 startup.sh。
   
   弹出下面页面说明启动成功：
   

项目集成

导入依赖

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

<dependency>
    <groupId>top.javatoolgroupId>
    <artifactId>canal-spring-boot-starterartifactId>
    <version>1.2.1-RELEASEversion>
dependency>
1
2
3
4
5

添加配置信息

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

监听模块的名字和目录名一致，在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> {
    @Override
    public void insert(User user) {
        log.info("User_insert：{}", user.toString());
    }

    @Override
    public void update(User before, User after) {
        log.info("User_update_before：{}", before.toString());
        log.info("User_update_after：{}", after.toString());
    }

    @Override
    public void delete(User user) {
        log.info("User_delete：{}", user.toString());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

测试

我们现在去操作数据库对表进行新增、修改、删除都会被监听到了。