使用 Debezium 在 Spring 应用程序中监听 MySQL binlog

要在 Spring 应用程序中监听 MySQL 的 binlog（二进制日志），您可以使用开源库 Debezium。Debezium 提供了一个 Kafka Connect 插件，用于捕获 MySQL、PostgreSQL、MongoDB 等数据库的数据变更事件。

以下是使用 Debezium 在 Spring 应用程序中监听 MySQL binlog 的步骤：

1. 在 pom.xml 文件中添加 Debezium 的依赖项：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-mysql</artifactId>
    <version>1.8.0.Final</version>
</dependency>

2. 配置 Debezium

在 application.yml 文件中添加以下配置

debezium:
  connector:
    name: my-connector
    connector.class: io.debezium.connector.mysql.MySqlConnector
    tasks.max: 1
    database.hostname: localhost
    database.port: 3306
    database.user: root
    database.password: root
    database.server.id: 184054
    database.server.name: my-server
    database.history.kafka.bootstrap.servers: localhost:9092
    database.history.kafka.topic: dbhistory.my-server
    database.include.list: mydatabase

3. 创建一个配置类 BinlogListenerConfiguration用于设置 Debezium 的配置并启动嵌入式引擎：

import io.debezium.config.Configuration;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

@Configuration
 public class BinlogListenerConfiguration {
        @Bean
        public DebeziumEngine<ChangeEvent<Object, Object>> debeziumEngine() {
            // Debezium 配置
            Configuration config = Configuration.create()
                    .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                    .with("offset.storage", "org.apache.kafka.connect.storage.MemoryOffsetBackingStore")
                    .with("offset.flush.interval.ms", 60000)
                    .with("name", "mysql-connector")
                    .with("database.hostname", "localhost")
                    .with("database.port", 3306)
                    .with("database.user", "your_database_user")
                    .with("database.password", "your_database_password")
                    .with("database.server.id", 1)
                    .with("database.server.name", "your_database_server_name")
                    .with("database.whitelist", "your_database_name")
                    .with("database.history", "io.debezium.relational.history.MemoryDatabaseHistory")
                    .build();
            // 创建 Debezium 引擎
            DebeziumEngine<ChangeEvent<Object, Object>> engine = DebeziumEngine.create(ChangeEvent.class)
                    .using(config)
                    .notifying(record -> {
                        System.out.println("Received change event: " + record);
                    }).build();
            // 使用单独的线程启动 Debezium 引擎
            Executor executor = Executors.newSingleThreadExecutor();
            executor.execute(engine);
            return engine;

        }

    }

在这个配置类中，我们首先创建了一个 Debezium 配置对象，其中包含了连接到 MySQL 数据库所需的所有信息。然后，我们创建了一个 Debezium 引擎，该引擎在接收到 binlog 事件时调用指定的通知函数。最后，我们使用一个单独的线程启动 Debezium 引擎。

4. 现在，每当数据库发生更改时，您的 Spring 应用程序都会接收到 binlog 事件。

在上面的示例中，我们只是简单地将接收到的事件打印到控制台。您可以根据需要修改通知函数以处理这些事件，例如将它们发送到消息队列、更新缓存等。

请注意，Debezium 仅支持 InnoDB 存储引擎。确保您的 MySQL 数据库使用 InnoDB 存储引擎。此外，确保已启用 binlog，并将 binlog_format 设置为 ROW。要了解有关 Debezium 的更多信息，如何配置过滤器等，请访问官方文档：debezium.io/documentati…