Spring Boot 整合 Canal 和 RabbitMQ 以监听数据变更记录

2023-11-10 13:45:36

前言

在实际应用中，我们经常需要对数据库中的数据变更进行记录，以便进行后续处理，例如同步数据到其他系统、生成报表、或者触发业务逻辑。传统的方法是使用数据库的触发器或审计功能来记录数据变更，但这些方法往往与业务代码耦合度较高，不便于维护和扩展。

2. Canal 简介

Canal 是一个开源的 MySQL binlog 解析工具，它可以实时地从 MySQL 中捕获变更数据并将其发送到下游系统。Canal 使用了 Canal Server 和 Canal Client 的架构，Canal Server 负责监听 MySQL 的 binlog，并将其解析成变更事件，Canal Client 则负责从 Canal Server 接收变更事件并将其发送到下游系统。

3. RabbitMQ 简介

RabbitMQ 是一个开源的消息队列，它可以可靠地传递消息。RabbitMQ 使用了 Exchange、Queue 和 Binding 的架构，Exchange 用于接收和分发消息，Queue 用于存储消息，Binding 用于将 Exchange 和 Queue 绑定在一起，以便消息可以从 Exchange 路由到 Queue。

4. Spring Boot 集成 Canal 和 RabbitMQ

下面我们将介绍如何在 Spring Boot 项目中集成 Canal 和 RabbitMQ 来实现数据库变更记录。

4.1 添加依赖

首先，我们需要在项目中添加 Canal 和 RabbitMQ 的依赖：

<dependency>
    <groupId>org.apache.canal</groupId>
    <artifactId>canal-client</artifactId>
    <version>1.1.4</version>
</dependency>
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.13.3</version>
</dependency>

4.2 配置 Canal

接下来，我们需要配置 Canal。我们可以通过在 application.yml 文件中添加如下配置来配置 Canal：

canal:
  host: 127.0.0.1
  port: 3306
  username: root
  password: root
  destination: example
  filter:
    database: example
    table: user

其中：

host：MySQL 的主机地址
port：MySQL 的端口号
username：MySQL 的用户名
password：MySQL 的密码
destination：Canal Server 的目的地，用于标识该 Canal Client 实例
filter：Canal 的过滤规则，用于指定要监听的数据库和表

4.3 配置 RabbitMQ

接下来，我们需要配置 RabbitMQ。我们可以通过在 application.yml 文件中添加如下配置来配置 RabbitMQ：

rabbitmq:
  host: 127.0.0.1
  port: 5672
  username: guest
  password: guest
  virtual-host: /

其中：

host：RabbitMQ 的主机地址
port：RabbitMQ 的端口号
username：RabbitMQ 的用户名
password：RabbitMQ 的密码
virtual-host：RabbitMQ 的虚拟主机

4.4 创建监听器

接下来，我们需要创建一个监听器来监听 Canal 的变更事件并将其发送到 RabbitMQ。我们可以通过实现 CanalEventListener 接口来创建一个监听器，如下所示：

public class MyCanalEventListener implements CanalEventListener {

    @Override
    public void onEvent(CanalEvent event) {
        for (CanalEntry entry : event.getEntries()) {
            if (entry.getEntryType() == EntryType.ROWDATA) {
                RowData rowData = entry.getRowData();
                // 这里可以对 rowData 进行处理，例如将其发送到 RabbitMQ
            }
        }
    }

}

4.5 启动 Canal Client

最后，我们需要启动 Canal Client。我们可以通过在 Spring Boot 项目的主类中添加如下代码来启动 Canal Client：

public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
    CanalClient canalClient = new CanalClient();
    canalClient.start();
}