一、背景与选型

在微服务或数据同步场景中，我们经常需要实时捕获 MySQL 数据库的增删改操作，并触发后续业务逻辑（如刷新缓存、同步到 ES、发送消息等）。
常见的方案有：

• Canal（阿里开源）

• Debezium（基于 Kafka Connect，但也可嵌入式运行）

• Maxwell

本文选择Debezium Embedded Engine，因为它：

• 无需依赖 Kafka，可直接在 Spring Boot 应用中内嵌运行；

• 支持 MySQL、PostgreSQL、MongoDB 等多种数据库；

• 提供完整的变更事件结构（before/after、元数据）；

• 容错性好，支持断点续传（offset 存储）。

适用场景：中小型项目，希望快速集成 CDC（Change Data Capture）功能，又不希望引入额外中间件。

二、环境准备

2.1 软件版本

• JDK 17+（Spring Boot 3.x 要求）

• Spring Boot 3.x

• MySQL 5.7 或 8.0（需开启 binlog）

• Debezium 2.7.x

2.2 MySQL 开启 binlog

编辑 MySQL 配置文件（my.cnf或my.ini），添加以下内容：

[mysqld] log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL server_id = 1 # 确保唯一，不能与 Debezium 的 database.server.id 冲突

重启 MySQL 服务后，执行 SQL 验证：

SHOW VARIABLES LIKE 'log_bin'; -- 应为 ON SHOW VARIABLES LIKE 'binlog_format'; -- 应为 ROW

2.3 创建测试库表和 CDC 用户

CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%'; FLUSH PRIVILEGES; -- 创建测试库和表 CREATE DATABASE IF NOT EXISTS park; USE park; CREATE TABLE test_user ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(50), age INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

三、创建 Spring Boot 项目

使用 IDEA 或 Spring Initializr 创建一个 Spring Boot 项目，引入以下依赖（pom.xml）：

<properties> <java.version>17</java.version> <debezium.version>2.7.0.Final</debezium.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- Debezium 核心 API --> <dependency> <groupId>io.debezium</groupId> <artifactId>debezium-api</artifactId> <version>${debezium.version}</version> </dependency> <!-- Debezium 嵌入式引擎 --> <dependency> <groupId>io.debezium</groupId> <artifactId>debezium-embedded</artifactId> <version>${debezium.version}</version> </dependency> <!-- Debezium 文件存储（用于 offset 和 schema history） --> <dependency> <groupId>io.debezium</groupId> <artifactId>debezium-storage-file</artifactId> <version>${debezium.version}</version> </dependency> <!-- Debezium MySQL 连接器 --> <dependency> <groupId>io.debezium</groupId> <artifactId>debezium-connector-mysql</artifactId> <version>${debezium.version}</version> </dependency> </dependencies>

四、核心代码编写-可以写在yml里面

4.1 配置类（DebeziumConfig）

集中管理连接器配置，方便后续调整。

package com.example.demo.config; import org.springframework.context.annotation.Configuration; import java.util.Properties; @Configuration public class DebeziumConfig { public Properties getDebeziumProperties() { Properties props = new Properties(); // 连接器名称 props.setProperty("name", "mysql-connector"); props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector"); // 偏移量存储（记录消费进度） props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore"); props.setProperty("offset.storage.file.filename", "./offsets.dat"); props.setProperty("offset.flush.interval.ms", "60000"); // 数据库连接 props.setProperty("database.hostname", "localhost"); props.setProperty("database.port", "3306"); props.setProperty("database.user", "debezium"); props.setProperty("database.password", "dbz"); props.setProperty("database.server.id", "184054"); // 必须唯一 props.setProperty("topic.prefix", "dbserver"); // 事件主题前缀 // 过滤：只监听 park 库下的 test_user 表 props.setProperty("database.include.list", "park"); props.setProperty("table.include.list", "park.test_user"); // 时区与 SSL props.setProperty("database.connectionTimeZone", "UTC"); props.setProperty("database.use.ssl", "false"); props.setProperty("database.allowPublicKeyRetrieval", "true"); // Schema 历史存储（用于 DDL 变更跟踪） props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory"); props.setProperty("schema.history.internal.file.filename", "./dbhistory.dat"); // 快照模式：never 表示不执行初始快照，只监听增量 props.setProperty("snapshot.mode", "never"); return props; } }

参数说明：

• snapshot.mode=never：启动后不进行全表快照，只监听后续变更。若希望首次启动时先同步现有数据，可改为initial。

• offset.storage.file.filename：记录已消费的 binlog 位置，重启后从中断处继续。

• schema.history.internal.file.filename：记录表结构变化历史，用于正确解析事件中的字段。

4.2 监听器组件（DebeziumListener）

负责启动 Debezium 引擎，接收并解析变更事件。

package com.example.demo.listener; import com.example.demo.config.DebeziumConfig; import com.fasterxml.jackson.databind.ObjectMapper; import io.debezium.engine.ChangeEvent; import io.debezium.engine.DebeziumEngine; import io.debezium.engine.format.Json; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @Component public class DebeziumListener { private static final Logger LOG = LoggerFactory.getLogger(DebeziumListener.class); @Autowired private DebeziumConfig debeziumConfig; private final ExecutorService executor = Executors.newSingleThreadExecutor(); private DebeziumEngine<ChangeEvent<String, String>> engine; private final ObjectMapper objectMapper = new ObjectMapper(); @PostConstruct public void start() { var props = debeziumConfig.getDebeziumProperties(); this.engine = DebeziumEngine.create(Json.class) .using(props) .notifying(this::handleChangeEvent) .build(); executor.execute(() -> { try { engine.run(); } catch (Exception e) { LOG.error("Debezium engine runtime error: ", e); } }); LOG.info("Debezium Engine started asynchronously."); } /** * 处理每个变更事件 */ private void handleChangeEvent(ChangeEvent<String, String> event) { String value = event.value(); if (value == null) return; try { Map<String, Object> payload = objectMapper.readValue(value, Map.class); Map<String, Object> payloadData = (Map<String, Object>) payload.get("payload"); if (payloadData == null) return; // 获取源信息 Map<String, Object> source = (Map<String, Object>) payloadData.get("source"); String db = (String) source.get("db"); String table = (String) source.get("table"); // 操作类型：c=insert, u=update, d=delete String operation = (String) payloadData.get("op"); if (operation == null) { LOG.warn("Operation is null, skipping event for {}.{}", db, table); return; } Map<String, Object> before = (Map<String, Object>) payloadData.get("before"); Map<String, Object> after = (Map<String, Object>) payloadData.get("after"); switch (operation) { case "c" -> { LOG.info("插入数据 on {}.{}: {}", db, table, after); // TODO: 业务处理 } case "u" -> { LOG.info("更新数据 on {}.{}: before={}, after={}", db, table, before, after); // TODO: 业务处理 } case "d" -> { LOG.info("删除数据 on {}.{}: {}", db, table, before); // TODO: 业务处理 } default -> LOG.debug("Unknown operation: {}", operation); } } catch (IOException e) { LOG.error("Error parsing change event: {}", e.getMessage(), e); } } @PreDestroy public void stop() { if (engine != null) { try { engine.close(); } catch (Exception e) { LOG.error("Error closing Debezium engine", e); } } executor.shutdownNow(); LOG.info("Debezium Engine stopped."); } }

注意：在handleChangeEvent中，你可以注入 Service 层，将变更数据同步到 Redis、Elasticsearch 、Mqtt。

五、运行与验证

5.1 启动 Spring Boot 应用

启动主类，观察控制台输出：

Debezium Engine started asynchronously.

5.2 在 MySQL 中执行 DML 操作

-- 插入 INSERT INTO park.test_user (name, age) VALUES ('张三', 25); -- 更新 UPDATE park.test_user SET age = 26 WHERE name = '张三'; -- 删除 DELETE FROM park.test_user WHERE name = '张三';

5.3 查看应用日志

你会看到类似以下格式的事件日志：

六、进阶说明

6.1 数据格式详解

Debezium 输出的 JSON 结构大致如下：

{ "payload": { "before": { "id": 1, "name": "张三", "age": 25 }, "after": { "id": 1, "name": "张三", "age": 26 }, "source": { "db": "park", "table": "test_user", "server_id": 184054, "ts_ms": 1721212345678, "gtid": null, "file": "mysql-bin.000001", "pos": 1234 }, "op": "u", "ts_ms": 1721212345678 } }

• op：c插入，u更新，d删除，r表示快照（若开启）。

• before/after：分别为变更前/后的行数据（删除操作只有 before）。

6.2 断点续传原理

• offsets.dat文件记录了当前消费的 binlog 位置（文件名 + 偏移量）。

• 重启应用后，Debezium 会从该位置继续读取，不会丢失数据。

• 若想重新消费全部数据，只需删除offsets.dat和dbhistory.dat，并将snapshot.mode改为initial。

6.3 多表监听

修改table.include.list为多个表，用逗号分隔：

props.setProperty("table.include.list", "park.test_user, park.another_table");

也可以通过database.include.list监听的库，再通过table.exclude.list排除部分表。

七、常见问题及解决方法