Spring Boot 整合 Debezium 实现 MySQL 增量数据监听（嵌入式版）

一、背景与选型在微服务或数据同步场景中我们经常需要实时捕获 MySQL 数据库的增删改操作并触发后续业务逻辑如刷新缓存、同步到 ES、发送消息等。常见的方案有Canal阿里开源Debezium基于 Kafka Connect但也可嵌入式运行Maxwell本文选择Debezium Embedded Engine因为它无需依赖 Kafka可直接在 Spring Boot 应用中内嵌运行支持 MySQL、PostgreSQL、MongoDB 等多种数据库提供完整的变更事件结构before/after、元数据容错性好支持断点续传offset 存储。适用场景中小型项目希望快速集成 CDCChange Data Capture功能又不希望引入额外中间件。二、环境准备2.1 软件版本JDK 17Spring Boot 3.x 要求Spring Boot 3.xMySQL 5.7 或 8.0需开启 binlogDebezium 2.7.x2.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; -- 应为 ROW2.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.xmlproperties java.version17/java.version debezium.version2.7.0.Final/debezium.version /properties dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- Debezium 核心 API -- dependency groupIdio.debezium/groupId artifactIddebezium-api/artifactId version${debezium.version}/version /dependency !-- Debezium 嵌入式引擎 -- dependency groupIdio.debezium/groupId artifactIddebezium-embedded/artifactId version${debezium.version}/version /dependency !-- Debezium 文件存储用于 offset 和 schema history -- dependency groupIdio.debezium/groupId artifactIddebezium-storage-file/artifactId version${debezium.version}/version /dependency !-- Debezium MySQL 连接器 -- dependency groupIdio.debezium/groupId artifactIddebezium-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.modenever启动后不进行全表快照只监听后续变更。若希望首次启动时先同步现有数据可改为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 DebeziumEngineChangeEventString, 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(ChangeEventString, String event) { String value event.value(); if (value null) return; try { MapString, Object payload objectMapper.readValue(value, Map.class); MapString, Object payloadData (MapString, Object) payload.get(payload); if (payloadData null) return; // 获取源信息 MapString, Object source (MapString, Object) payloadData.get(source); String db (String) source.get(db); String table (String) source.get(table); // 操作类型cinsert, uupdate, ddelete String operation (String) payloadData.get(op); if (operation null) { LOG.warn(Operation is null, skipping event for {}.{}, db, table); return; } MapString, Object before (MapString, Object) payloadData.get(before); MapString, Object after (MapString, 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 } }opc插入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排除部分表。七、常见问题及解决方法问题现象可能原因解决方案启动时连接 MySQL 失败用户权限不足 / SSL 问题检查用户授权添加database.allowPublicKeyRetrievaltrue无任何变更事件输出binlog 未开启或格式不是 ROW检查 MySQL 配置确认binlog_formatROW事件中 before/after 为 nullbinlog_row_image不是 FULL设置为FULL并重启 MySQL重启后重复消费或漏消费offset 文件损坏删除offsets.dat和dbhistory.dat设置snapshot.modeinitial重新同步解析 JSON 异常表结构变更未正确记录确保schema.history.internal.file.filename文件持久化不要删除database.server.id冲突与 MySQL 的 server_id 或其它连接器重复修改为不同的整数值