SpringBoot整合Canal实现数据库实时同步

前言

在微服务架构盛行的今天，数据一致性成为了一个关键挑战。当业务数据在MySQL中发生变化时，如何实时同步到其他服务或缓存中？阿里巴巴开源的Canal组件为我们提供了完美的解决方案。今天，我将带你深入探索SpringBoot整合Canal的技术内幕，让你轻松掌握这一核心技术。

一、什么是Canal？

Canal是阿里巴巴开源的一个基于MySQL数据库增量日志解析的组件，它模拟MySQL主从复制的交互协议，伪装成MySQL的从节点，向MySQL主节点发送dump协议，获取到MySQL的二进制日志(binlog)后，再解析为便于理解和使用的数据格式。

1.1 Canal工作原理

MySQL主库(master)
↓ (binlog日志)
Canal Server (模拟slave)
↓ (解析后的数据变更事件)
Canal Client/SpringBoot应用
↓ (业务处理)
下游系统(Redis/ES/MQ等)

Canal的核心原理就是：

二、环境准备

2.1 MySQL配置

首先，确保MySQL已开启binlog功能，并设置为ROW模式：

— 查看binlog配置
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

— 如果未开启，需在my.cnf中添加以下配置
[mysqld]
# 开启binlog
log–bin=mysql–bin
# binlog格式必须为ROW
binlog–format=ROW
# 服务器ID，唯一标识
server–id=1
# binlog过期时间（天）
expire_logs_days=7
# 单个binlog文件大小
max_binlog_size=500M

2.2 创建Canal专用用户

创建Canal用户并授权：

— 创建canal用户
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';

— 授予权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

— 刷新权限
FLUSH PRIVILEGES;

注意：必须授予REPLICATION SLAVE和REPLICATION CLIENT权限，Canal才能读取binlog。

2.3 部署Canal Server

下载Canal Server

从GitHub下载最新稳定版（推荐1.1.7版本）：

wget https://github.com/alibaba/canal/releases/download/canal-1.1.7/canal.deployer-1.1.7.tar.gz

# 解压
mkdir -p /opt/canal
tar -zxvf canal.deployer-1.1.7.tar.gz -C /opt/canal

配置Canal Server

修改conf/example/instance.properties配置文件：

# MySQL连接信息
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset=UTF-8

# 监听的表（正则表达式）
# .*\\\\..* 表示监听所有库的所有表
# 可以指定为具体的表，如：test_db\\\\.user_info
canal.instance.filter.regex=.*\\\\..*

# 表黑名单
canal.instance.filter.black.regex=mysql\\\\.information_schema.*

# slaveId（需要唯一）
canal.instance.mysql.slaveId=1234

修改conf/canal.properties全局配置：

# Canal服务端口
canal.port=11111
canal.metrics.pull.port=11112

# 实例列表
canal.destinations=example

# 管理端口
canal.admin.manager.port=8089

启动Canal Server

# 启动
sh bin/startup.sh

# 查看日志
tail -f logs/canal/canal.log
tail -f logs/example/example.log

# 停止
sh bin/stop.sh

启动成功后，你会看到类似以下日志：

2026-02-02 17:00:00.123 [main] INFO com.alibaba.otter.canal.instance.core.CanalInstanceWithManager – [example] init successful

三、SpringBoot集成Canal

3.1 创建SpringBoot项目

创建一个新的SpringBoot项目，添加必要的依赖：

<?xml version="1.0" encoding="UTF-8"?>
<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>

<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.18</version>
<relativePath/>
</parent>

<groupId>com.imooc</groupId>
<artifactId>canal-sync-demo</artifactId>
<version>1.0.0</version>

<properties>
<java.version>8</java.version>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>

<dependencies>
<!– Spring Boot Web –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>

<!– Canal Spring Boot Starter –>
<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>

<!– JPA注解支持 –>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>persistence-api</artifactId>
<version>1.0</version>
</dependency>

<!– Lombok –>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>

<!– Redis –>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

<!– MyBatis Plus –>
<dependency>
<groupId>com.baomidou</groupId>
<artifactId>mybatis-plus-boot-starter</artifactId>
<version>3.5.5</version>
</dependency>

<!– MySQL驱动 –>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<scope>runtime</scope>
</dependency>

<!– fastjson –>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

3.2 配置文件

创建application.yml配置文件：

server:
port: 8080

spring:
application:
name: canal–sync–demo

# MySQL配置
datasource:
url: jdbc:mysql://127.0.0.1:3306/test_db?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai
username: root
password: 123456
driver-class-name: com.mysql.cj.jdbc.Driver

# Redis配置
redis:
host: 127.0.0.1
port: 6379
password:
database: 0

# Canal配置
canal:
# Canal Server地址
server: 127.0.0.1:11111
# 实例名称（对应Canal Server配置的destination）
destination: example
# Canal Server用户名和密码（如果Canal Server配置了认证）
user-name: canal
password: canal

# 日志配置
logging:
level:
top.javatool.canal.client: warn

注意：top.javatool.canal.client的日志级别设置为warn，避免Canal客户端的日志过多。

3.3 创建实体类

假设我们有一个数据字典表data_dictionary，对应的实体类如下：

package com.xxx.pojo;

import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import javax.persistence.Column;
import javax.persistence.Id;

/**
* 数据字典实体类
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
@TableName("data_dictionary")
public class DataDictionary {

/**
* 主键ID
*/
@Id
@TableId
@Column(name = "id")
private String id;

/**
* 类型编码
*/
@Column(name = "type_code")
private String typeCode;

/**
* 类型名称
*/
@Column(name = "type_name")
private String typeName;

/**
* 字典项键
*/
@Column(name = "item_key")
private String itemKey;

/**
* 字典项值
*/
@Column(name = "item_value")
private String itemValue;

/**
* 排序
*/
@Column(name = "sort")
private Integer sort;

/**
* 图标
*/
@Column(name = "icon")
private String icon;

/**
* 是否启用
*/
@Column(name = "enable")
private Boolean enable;
}

关键点说明：

• @TableName("data_dictionary")：指定对应的数据库表名
• @Id和@Column：这些注解来自JPA，Canal会根据这些注解将数据库字段映射到实体类属性
• 类名、属性名可以与表名、字段名不同，通过注解映射

3.4 创建Canal监听处理器

这是最核心的部分！我们需要创建一个类来实现EntryHandler<T>接口：

package com.xxx.canal;

import com.alibaba.fastjson.JSON;
import com.xxx.pojo.DataDictionary;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

import javax.annotation.Resource;

/**
* 数据字典同步处理器
* 监听data_dictionary表的数据变更，实时同步到Redis
*/
@Slf4j
@Component
@CanalTable("data_dictionary") // 指定监听的表名
public class DataDictSyncHandler implements EntryHandler<DataDictionary> {

@Resource
private RedisTemplate<String, Object> redisTemplate;

/**
* 监听新增操作
* 当MySQL中执行INSERT时，此方法会被回调
*
* @param dataDictionary 新增的数据
*/
@Override
public void insert(DataDictionary dataDictionary) {
log.info("=== Canal监听到新增操作 ===");
log.info("数据内容：{}", JSON.toJSONString(dataDictionary));

// 业务逻辑：将数据同步到Redis
syncToRedis(dataDictionary);

log.info("新增操作处理完成");
}

/**
* 监听更新操作
* 当MySQL中执行UPDATE时，此方法会被回调
*
* @param before 更新前的数据（老数据）
* @param after 更新后的数据（新数据）
*/
@Override
public void update(DataDictionary before, DataDictionary after) {
log.info("=== Canal监听到更新操作 ===");
log.info("更新前数据：{}", JSON.toJSONString(before));
log.info("更新后数据：{}", JSON.toJSONString(after));

// 业务逻辑：删除旧缓存，写入新缓存
deleteFromRedis(before.getId());
syncToRedis(after);

log.info("更新操作处理完成");
}

/**
* 监听删除操作
* 当MySQL中执行DELETE时，此方法会被回调
*
* @param dataDictionary 被删除的数据
*/
@Override
public void delete(DataDictionary dataDictionary) {
log.info("=== Canal监听到删除操作 ===");
log.info("被删除数据：{}", JSON.toJSONString(dataDictionary));

// 业务逻辑：从Redis中删除对应数据
deleteFromRedis(dataDictionary.getId());

log.info("删除操作处理完成");
}

/**
* 将数据同步到Redis
*
* @param dataDictionary 数据字典对象
*/
private void syncToRedis(DataDictionary dataDictionary) {
try {
String key = "data_dict:" + dataDictionary.getId();
redisTemplate.opsForValue().set(key, dataDictionary);
log.info("数据已同步到Redis，key：{}", key);
} catch (Exception e) {
log.error("同步数据到Redis失败：", e);
}
}

/**
* 从Redis中删除数据
*
* @param id 数据ID
*/
private void deleteFromRedis(String id) {
try {
String key = "data_dict:" + id;
redisTemplate.delete(key);
log.info("已从Redis删除数据，key：{}", key);
} catch (Exception e) {
log.error("从Redis删除数据失败：", e);
}
}
}

核心要点：

3.5 启动类

创建SpringBoot启动类：

package com.xxx;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
* Canal同步演示启动类
*/
@SpringBootApplication
public class CanalSyncApplication {

public static void main(String[] args) {
SpringApplication.run(CanalSyncApplication.class, args);
}
}

四、测试验证

4.1 准备测试数据

在MySQL中创建测试表并插入数据：

— 创建数据库
CREATE DATABASE IF NOT EXISTS test_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

USE test_db;

— 创建数据字典表
CREATE TABLE IF NOT EXISTS `data_dictionary` (
`id` varchar(64) NOT NULL COMMENT '主键ID',
`type_code` varchar(50) NOT NULL COMMENT '类型编码',
`type_name` varchar(100) DEFAULT NULL COMMENT '类型名称',
`item_key` varchar(50) DEFAULT NULL COMMENT '字典项键',
`item_value` varchar(200) DEFAULT NULL COMMENT '字典项值',
`sort` int(11) DEFAULT '0' COMMENT '排序',
`icon` varchar(100) DEFAULT NULL COMMENT '图标',
`enable` tinyint(1) DEFAULT '1' COMMENT '是否启用',
PRIMARY KEY (`id`),
KEY `idx_type_code` (`type_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据字典表';

— 插入测试数据
INSERT INTO `data_dictionary`
VALUES ('1', 'gender', '性别', 'male', '男', 1, 'icon-male', 1),
('2', 'gender', '性别', 'female', '女', 2, 'icon-female', 1);

4.2 启动应用

4.3 测试数据同步

测试新增

在MySQL中执行INSERT：

INSERT INTO `data_dictionary`
VALUES ('3', 'status', '状态', 'active', '激活', 1, NULL, 1);

观察应用日志：

=== Canal监听到新增操作 ===
数据内容：{"id":"3","typeCode":"status","typeName":"状态","itemKey":"active","itemValue":"激活","sort":1,"icon":null,"enable":true}
数据已同步到Redis，key：data_dict:3
新增操作处理完成

检查Redis中是否有对应数据：

redis-cli
127.0.0.1:6379> GET data_dict:3
{"id":"3","typeCode":"status","typeName":"状态","itemKey":"active","itemValue":"激活","sort":1,"icon":null,"enable":true}

测试更新

在MySQL中执行UPDATE：

UPDATE `data_dictionary`
SET `item_value`='启用中', `sort`=2
WHERE `id`='3';

观察应用日志：

=== Canal监听到更新操作 ===
更新前数据：{"id":"3","typeCode":"status","typeName":"状态","itemKey":"active","itemValue":"激活","sort":1,"icon":null,"enable":true}
更新后数据：{"id":"3","typeCode":"status","typeName":"状态","itemKey":"active","itemValue":"启用中","sort":2,"icon":null,"enable":true}
已从Redis删除数据，key：data_dict:3
数据已同步到Redis，key：data_dict:3
更新操作处理完成

测试删除

在MySQL中执行DELETE：

DELETE FROM `data_dictionary` WHERE `id`='3';

观察应用日志：

=== Canal监听到删除操作 ===
被删除数据：{"id":"3","typeCode":"status","typeName":"状态","itemKey":"active","itemValue":"启用中","sort":2,"icon":null,"enable":true}
已从Redis删除数据，key：data_dict:3
删除操作处理完成

检查Redis中数据是否已被删除：

127.0.0.1:6379> GET data_dict:3
(nil)

五、高级应用场景

5.1 多表监听

如果需要监听多张表，可以创建多个Handler：

// 用户表监听器
@CanalTable("t_user")
@Component
public class UserHandler implements EntryHandler<User> {
@Override
public void insert(User user) {
// 处理用户新增
}

@Override
public void update(User before, User after) {
// 处理用户更新
}

@Override
public void delete(User user) {
// 处理用户删除
}
}

// 订单表监听器
@CanalTable("t_order")
@Component
public class OrderHandler implements EntryHandler<Order> {
@Override
public void insert(Order order) {
// 处理订单新增
}

@Override
public void update(Order before, Order after) {
// 处理订单更新
}

@Override
public void delete(Order order) {
// 处理订单删除
}
}

5.2 同步到Elasticsearch

@CanalTable("t_product")
@Component
@Slf4j
public class ProductEsHandler implements EntryHandler<Product> {

@Autowired
private ElasticsearchRestTemplate esTemplate;

@Override
public void insert(Product product) {
// 同步到ES
esTemplate.save(product);
}

@Override
public void update(Product before, Product after) {
// 更新ES文档
esTemplate.save(after);
}

@Override
public void delete(Product product) {
// 从ES删除文档
esTemplate.delete(product.getId(), "product");
}
}

5.3 发送消息到MQ

@CanalTable("t_order")
@Component
@Slf4j
public class OrderMqHandler implements EntryHandler<Order> {

@Autowired
private RabbitTemplate rabbitTemplate;

@Override
public void insert(Order order) {
// 发送订单创建消息
rabbitTemplate.convertAndSend("order.exchange", "order.created", order);
}

@Override
public void update(Order before, Order after) {
// 判断订单状态变化，发送相应消息
if (!before.getStatus().equals(after.getStatus())) {
rabbitTemplate.convertAndSend("order.exchange", "order.status.changed", after);
}
}

@Override
public void delete(Order order) {
// 发送订单删除消息
rabbitTemplate.convertAndSend("order.exchange", "order.deleted", order);
}
}

六、生产环境优化建议

6.1 性能优化

canal:
batch-size: 1000 # 每次拉取1000条

@Async
@Override
public void insert(DataDictionary dataDictionary) {
// 异步处理
}

# 在Canal Server配置中精简监听表
canal.instance.filter.regex=test_db\\\\.data_dictionary,test_db\\\\.t_user

6.2 高可用方案

6.3 监控告警

七、常见问题与解决方案

7.1 监听不到数据变更

可能原因：

解决方案：

— 检查binlog
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

— 检查用户权限
SHOW GRANTS FOR 'canal'@'%';

7.2 字段映射失败

问题描述：某些字段无法映射到实体类

解决方案：

@Column(name = "type_code") // 显式指定字段名
private String typeCode;

7.3 数据同步延迟

可能原因：

解决方案：

7.4 重复消费数据

问题描述：重启应用后，部分数据被重复处理

解决方案：

@Override
public void insert(DataDictionary dataDictionary) {
// 幂等性校验
String key = "processed:" + dataDictionary.getId();
if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
log.info("数据已处理过，跳过：{}", dataDictionary.getId());
return;
}

// 处理业务逻辑
syncToRedis(dataDictionary);

// 标记已处理（设置过期时间）
redisTemplate.opsForValue().set(key, "1", 24, TimeUnit.HOURS);
}

八、总结

通过本文的详细介绍，相信你已经掌握了SpringBoot整合Canal的核心技术。让我们总结一下关键要点：

核心优势

最佳实践

适用场景

• 缓存实时更新（Redis、Memcached）
• 搜索引擎数据同步（Elasticsearch）
• 数据归档与备份
• 多数据源同步
• 业务解耦（通过MQ发送变更事件）