SpringBoot整合 Canal、RabbitMQ 监听数据变更

需求

我想要在SpringBoot中采用一种与业务代码解耦合的方式，来实现数据的变更记录，记录的内容是新数据，如果是更新操作还得有旧数据内容。

经过调研发现，使用Canal来监听MySQL的binlog变化可以实现这个需求，可是在监听到变化后需要马上保存变更记录，除非再做一些逻辑处理，于是我又结合了RabbitMQ来处理保存变更记录的操作。

步骤

启动MySQL环境，并开启binlog

启动Canal环境，为其创建一个MySQL账号，然后以Slave的形式连接MySQL

Canal服务模式设为TCP，用Java编写客户端代码，监听MySQL的binlog修改

Canal服务模式设为RabbitMQ，启动RabbitMQ环境，配置Canal和RabbitMQ的连接，用消息队列去接收binlog修改事件

环境搭建

环境搭建基于docker-compose：

我们需要修改下Canal环境的配置文件：和，映射Canal中的以下两个路径：

配置文件中，意思是server上部署的instance列表，

这里的/example是指instance即实例名，要和上面内instance配置对应，canal会为实例创建对应的文件夹，一个Client对应一个实例

以下是我们需要准备的两个配置文件具体内容：

canal.properties

此时，即TCP直连，我们先开启这个服务，然后手写Java客户端代码去连接它，等下再改为RabbitMQ。

通过注释可以看到，canal支持的服务模式有：tcp、kafka、rocketMQ、rabbitMQ、pulsarMQ，即主流的消息队列都支持。

instance.properties

把这两个配置文件映射好，再次提醒，注意实例的路径名，默认是：

修改canal配置文件

我们需要修改这个实例配置文件，去连接MySQL，确保以下的配置正确：

mymysql是同为docker容器的MySQL环境，端口3306是指内部端口。

这里多说明一下，docker端口配置时假设为：13306:3306，那么容器对外的端口就是13306，内部是3306，在本示例中，MySQL和Canal都是容器环境，所以Canal连接MySQL需要满足以下条件：

处于同一网段（docker-compose.yml中的mynetwork）

访问内部端口（即3306，而非13306）

dbUsername和dbPassword为MySQL账号密码，为了开发方便可以使用，但是我仍建议自行创建用户并分配访问权限：

用navicat或者shell去登录canal这个用户，可以访问即创建成功

整合SpringBoot Canal实现客户端

Maven依赖：

新增组件并启动：

启动类Application：

启动程序，此时新增或修改数据库中的数据，我们就能从客户端中监听到

不过我建议监听的信息放到消息队列中，在空闲的时候去处理，所以直接配置Canal整合RabbitMQ更好。

Canal整合RabbitMQ

修改canal.properties中的serverMode：

修改instance.properties中的topic：

然后找到关于RabbitMQ的配置：

重新启动容器，进入RabbitMQ管理页面创建exchange交换器和队列queue：

新建exchange，命名为：

新建queue，命名为：

绑定exchange和queue，设置为：，这里对应上面instance.properties的

顺带一提，上面这段可以忽略，因为在SpringBoot的RabbitMQ配置中，会自动创建交换器exchange和队列queue，不过手动创建的话，可以在忽略SpringBoot的基础上，直接在RabbitMQ的管理页面上看到修改记录的消息。

SpringBoot整合RabbitMQ

依赖：

application.yml：

RabbitMQ配置类：

Canal消息生产者：

Canal消息消费者：

测试一下，修改MySQL中的一条消息，Canal就会发送信息到RabbitMQ，我们就能从监听的RabbitMQ队列中得到该条消息。