你是否也曾为记录数据变更而烦恼？如果直接在业务代码中埋点，不仅耦合度高，而且难以统一管理。今天，我们介绍一种与业务代码解耦的方案：利用Canal监听MySQL的Binlog，再通过RabbitMQ异步处理变更消息，最终在SpringBoot项目中落地实现。
核心需求与实现思路
我需要在SpringBoot项目中，以一种低侵入的方式记录数据变更。记录内容要包含新数据，如果是更新操作，还得能获取到旧数据。经过调研，发现通过Canal监听MySQL的Binlog可以完美满足需求。
但直接处理变更逻辑可能不够灵活，因此我引入了RabbitMQ作为消息中间件。Canal监听到变化后，将事件投递到RabbitMQ，再由消费者异步处理保存记录等逻辑，实现了监听与处理的解耦。
整体实现步骤
- 启动MySQL环境，并开启Binlog日志。
- 启动Canal服务，为其配置一个具有复制权限的MySQL账号，使其能以Slave身份连接MySQL。
- Canal以TCP模式运行，编写Java客户端代码连接Canal，直接接收Binlog变更事件。
- Canal切换为RabbitMQ模式，启动RabbitMQ环境，配置Canal与RabbitMQ的连接，让变更事件通过消息队列传递。
- SpringBoot整合RabbitMQ，编写消费者处理来自Canal的变更消息。
环境搭建（基于Docker Compose）
我们使用Docker Compose一键拉起MySQL、RabbitMQ和Canal-Server服务。
version: "3"
services:
mysql:
network_mode: mynetwork
container_name: mymysql
ports:
- 3306:3306
restart: always
volumes:
- /etc/localtime:/etc/localtime
- /home/mycontainers/mymysql/data:/data
- /home/mycontainers/mymysql/mysql:/var/lib/mysql
- /home/mycontainers/mymysql/conf:/etc/mysql
environment:
- MYSQL_ROOT_PASSWORD=root
command:
--character-set-server=utf8mb4
--collation-server=utf8mb4_unicode_ci
--log-bin=/var/lib/mysql/mysql-bin
--server-id=1
--binlog-format=ROW
--expire_logs_days=7
--max_binlog_size=500M
image: mysql:5.7.20
rabbitmq:
container_name: myrabbit
ports:
- 15672:15672
- 5672:5672
restart: always
volumes:
- /etc/localtime:/etc/localtime
- /home/mycontainers/myrabbit/rabbitmq:/var/lib/rabbitmq
network_mode: mynetwork
environment:
- RABBITMQ_DEFAULT_USER=admin
- RABBITMQ_DEFAULT_PASS=123456
image: rabbitmq:3.8-management
canal-server:
container_name: canal-server
restart: always
ports:
- 11110:11110
- 11111:11111
- 11112:11112
volumes:
- /home/mycontainers/canal-server/conf/canal.properties:/home/admin/canal-server/conf/canal.properties
- /home/mycontainers/canal-server/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties
- /home/mycontainers/canal-server/logs:/home/admin/canal-server/logs
network_mode: mynetwork
depends_on:
- mysql
- rabbitmq
image: canal/canal-server:v1.1.5
我们需要重点关注Canal的配置文件，并通过卷映射(volumes)挂载到容器中。主要涉及两个文件：
/home/admin/canal-server/conf/canal.properties：Canal服务端全局配置。/home/admin/canal-server/conf/example/instance.properties：Canal实例配置，路径中的example是实例名，需与canal.properties中的配置对应。
关键配置文件详解
1. canal.properties
这是服务端主配置。我们首先以TCP模式运行，因此设置 canal.serverMode = tcp。注意文件末尾的RabbitMQ连接配置，稍后会用到。
################################################
######## common argument ############
################################################
# tcp bind ip
canal.ip =
# register ip to zookeeper
canal.register.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
...
# tcp, kafka, rocketMQ, rabbitMQ, pulsarMQ
canal.serverMode = tcp
...
################################################
######## destinations ############
################################################
canal.destinations = canal-exchange
# conf root dir
canal.conf.dir = ../conf
# auto scan instance dir add/remove and start/stop instance
canal.auto.scan = true
canal.auto.scan.interval = 5
...
#################################################
######## RabbitMQ ############
#################################################
rabbitmq.host = myrabbit
rabbitmq.virtual.host = /
rabbitmq.exchange = canal-exchange
rabbitmq.username = admin
rabbitmq.password = 123456
rabbitmq.deliveryMode =
提示：从注释可见，Canal支持tcp， kafka， rocketMQ， rabbitMQ， pulsarMQ等多种服务模式。
2. instance.properties
这是数据同步实例的配置，核心是配置其要连接的MySQL源。
################################################
# position info
canal.instance.master.address=mymysql:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
# table regex
canal.instance.filter.regex=.*\..*
# mq config
canal.mq.topic=canal-routing-key
canal.mq.partition=0
关键配置项说明：
canal.instance.master.address：MySQL地址。因为Canal和MySQL都在Docker的mynetwork网络内，所以这里直接使用容器名mymysql和内部端口3306。canal.instance.dbUsername/password：Canal连接MySQL使用的账号密码。
3. 创建Canal专用MySQL用户
为了安全，不建议直接使用root账号。我们需要在MySQL中创建一个专供Canal使用的用户。
# 进入MySQL容器
docker exec -it mymysql bash
# 登录MySQL
mysql -uroot -proot
-- 执行SQL
use mysql;
-- 创建用户 canal，密码 canal
create user 'canal'@'%' identified by 'canal';
-- 授予必要权限（主要是复制相关权限）
grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%';
flush privileges;
创建完成后，可以使用canal/canal进行连接测试。
阶段一：SpringBoot整合Canal（TCP直连模式）
首先，我们以TCP模式启动Canal，并在SpringBoot中编写客户端直接连接。
1. 添加Maven依赖
<canal.version>1.1.5</canal.version>
<!--canal-->
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>${canal.version}</version>
</dependency>
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.protocol</artifactId>
<version>${canal.version}</version>
</dependency>
2. 编写Canal客户端组件
这个组件负责连接Canal服务端，订阅数据库变更，并循环拉取、处理消息。
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.List;
@Component
public class CanalClient {
private final static int BATCH_SIZE = 1000;
public void run() {
// 创建链接，连接到localhost:11111，实例名为canal-exchange
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("localhost", 11111), "canal-exchange", "canal", "canal");
try {
//打开连接
connector.connect();
//订阅所有表
connector.subscribe(".*\\..*");
//回滚到未进行ack的地方
connector.rollback();
while (true) {
// 获取指定数量的数据（无确认）
Message message = connector.getWithoutAck(BATCH_SIZE);
long batchId = message.getId();
int size = message.getEntries().size();
//如果没有数据
if (batchId == -1 || size == 0) {
try {
//线程休眠2秒
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
//如果有数据，处理数据
printEntry(message.getEntries());
}
// 确认此batchId的消息已处理
connector.ack(batchId);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
connector.disconnect();
}
}
/**
* 打印canal server解析binlog获得的实体类信息
*/
private static void printEntry(List<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
//跳过事务开始/结束的实体类型
continue;
}
//RowChange对象，包含了一行数据变化的所有特征
CanalEntry.RowChange rowChage;
try {
rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR # parser of eromanga-event has an error , data:" + entry.toString(), e);
}
//获取操作类型：insert/update/delete类型
CanalEntry.EventType eventType = rowChage.getEventType();
//打印Header信息
System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
//判断是否是DDL语句
if (rowChage.getIsDdl()) {
System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
}
//获取RowChange对象里的每一行数据，打印出来
for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
//如果是删除语句
if (eventType == CanalEntry.EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
//如果是新增语句
} else if (eventType == CanalEntry.EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
//如果是更新的语句
} else {
//变更前的数据
System.out.println("------->; before");
printColumn(rowData.getBeforeColumnsList());
//变更后的数据
System.out.println("------->; after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<CanalEntry.Column> columns) {
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
3. 在启动类中运行客户端
@SpringBootApplication
public class BaseApplication implements CommandLineRunner {
@Autowired
private CanalClient canalClient;
@Override
public void run(String... args) throws Exception {
canalClient.run();
}
}
启动SpringBoot应用，此时在MySQL中执行增删改操作，控制台就会打印出详细的变更信息。但这只是第一步，接下来我们将引入RabbitMQ，让架构更优雅。
阶段二：Canal整合RabbitMQ
现在我们改造Canal，让它将变更事件发送到RabbitMQ，而非直接由TCP客户端处理。
1. 修改Canal配置
2. 重启Canal-Server容器
配置修改完成后，重启Canal-Server容器使配置生效。
3. （可选）在RabbitMQ管理页面创建交换机和队列
虽然SpringBoot可以自动声明，但手动创建有助于理解流程。
- 创建名为
canal-exchange 的交换机（Direct类型）。
- 创建名为
canal-queue 的队列。
- 将队列绑定到交换机，并设置
routing-key 为 canal-routing-key。
阶段三：SpringBoot整合RabbitMQ消费消息
现在，我们将编写SpringBoot应用作为RabbitMQ的消费者，处理来自Canal的变更消息。
1. 添加依赖
<amqp.version>2.3.4.RELEASE</amqp.version>
<!--消息队列-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>${amqp.version}</version>
</dependency>
2. 配置RabbitMQ连接（application.yml）
spring:
rabbitmq:
host: 192.168.0.108 # 你的RabbitMQ服务器IP
port: 5672
username: admin
password: 123456
# 消息确认配置
publisher-confirm-type: correlated
publisher-returns: true
3. RabbitMQ配置类
@Configuration
public class RabbitConfig {
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate();
template.setConnectionFactory(connectionFactory);
template.setMessageConverter(new Jackson2JsonMessageConverter());
return template;
}
/**
* 配置Listener容器工厂，解决消息转换问题
*/
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}
}
4. 声明队列、交换机及绑定（生产者配置）
public class RabbitConstant {
public static final String CanalQueue = "canal-queue";
public static final String CanalExchange = "canal-exchange";
public static final String CanalRouting = "canal-routing-key";
}
@Configuration
public class CanalProvider {
/**
* 声明队列
*/
@Bean
public Queue canalQueue() {
return new Queue(RabbitConstant.CanalQueue, true);
}
/**
* 声明直连交换机
*/
@Bean
DirectExchange canalExchange() {
return new DirectExchange(RabbitConstant.CanalExchange, true, false);
}
/**
* 绑定队列到交换机
*/
@Bean
Binding bindingCanal() {
return BindingBuilder.bind(canalQueue()).to(canalExchange()).with(RabbitConstant.CanalRouting);
}
}
5. 消息消费者
这是最终处理数据变更逻辑的地方。
/**
* Canal消息消费者
*/
@Component
@RabbitListener(queues = RabbitConstant.CanalQueue)
public class CanalComsumer {
private final SysBackupService sysBackupService;
public CanalComsumer(SysBackupService sysBackupService) {
this.sysBackupService = sysBackupService;
}
@RabbitHandler
public void process(Map<String, Object> msg) {
System.out.println("收到canal消息：" + msg);
boolean isDdl = (boolean) msg.get("isDdl");
// 不处理DDL事件
if (isDdl) {
return;
}
// 消息ID
int tid = (int) msg.get("id");
// 时间戳
long ts = (long) msg.get("ts");
// 数据库名
String database = (String) msg.get("database");
// 表名
String table = (String) msg.get("table");
// 操作类型：INSERT/UPDATE/DELETE
String type = (String) msg.get("type");
// 新数据
List<?> data = (List<?>) msg.get("data");
// 旧数据（仅UPDATE操作有值）
List<?> old = (List<?>) msg.get("old");
// 示例：跳过备份表自身产生的变更，防止无限循环
if ("sys_backup".equalsIgnoreCase(table)) {
return;
}
// 只处理需要的操作类型
if (!"INSERT".equalsIgnoreCase(type)
&& !"UPDATE".equalsIgnoreCase(type)
&& !"DELETE".equalsIgnoreCase(type)) {
return;
}
// TODO: 在这里编写你的业务逻辑，例如将 data 和 old 保存到另一张记录表
// sysBackupService.saveBackupRecord(...);
}
}
测试与总结
完成以上所有步骤后，启动整个系统设计：
- 运行
docker-compose up 启动基础设施。
- 启动你的SpringBoot应用。
- 在MySQL中任意修改一条数据。
- 观察SpringBoot应用的控制台，将会打印出从RabbitMQ接收到的、格式化的JSON变更消息。
至此，一个基于SpringBoot、Canal和RabbitMQ的数据库变更监听与异步处理方案就完整实现了。这套方案将数据变更的捕获与业务处理完全解耦，扩展性强，非常适合用于数据同步、审计日志、缓存更新等场景。如果你在实践过程中遇到问题，欢迎到云栈社区与大家交流探讨。
发表于 1 小时前
|
查看: 3|
回复: 0
