本文深入探讨了debezium在java应用中捕获mysql数据变更，特别是删除事件时常见的配置问题。核心内容包括正确配置`database.include.list`和`table.include.list`以精准指定监控范围，以及识别并替换已弃用的`database.whitelist`等属性。通过优化配置示例和注意事项，旨在帮助开发者构建稳定可靠的debezium cdc解决方案。
Debezium作为一个强大的分布式变更数据捕获（CDC）平台，能够将数据库的每一次数据变更事件流式传输到消息队列中。然而，在实际应用中，尤其是在Java集成场景下，开发者可能会遇到Debezium无法捕获预期数据变更（例如删除操作）的问题。这通常源于MySQL连接器配置不当。本教程将详细解析Debezium MySQL连接器的关键配置，并指出常见的配置陷阱及其解决方案。
Debezium引擎是Java应用程序中集成Debezium的核心组件。它通过DebeziumEngine.create()方法构建，并加载特定的连接器配置来监控数据库。对于MySQL，我们使用io.debezium.connector.mysql.MySqlConnector。
一个典型的Debezium引擎初始化流程如下：
public DebeziumSignal connect(Connection data) {
// 加载配置
final Configuration configuration = DebeziumConfigLoader.load(data);
// 构建Debezium引擎
engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
.using(configuration.asProperties()) // 使用配置属性
.notifying(this::handleEvent) // 设置事件处理器
.build();
// 返回一个信号对象，通常用于启动和停止引擎
return new DebeziumSignal(engine);
}其中，DebeziumConfigLoader.load(data)方法负责生成Debezium连接器所需的配置。
在配置Debezium MySQL连接器时，以下属性至关重要，并且是导致捕获失败的常见原因：
Debezium提供了精细的控制来指定要监控的数据库和表。理解并正确使用这些属性是确保数据变更被捕获的关键。
常见错误分析: 原始配置中可能存在将mysql.getTables()返回的表名列表直接赋给database.include.list的情况。这导致Debezium无法识别这些作为数据库名的表名，从而无法正确初始化监控。
Debezium在不同版本迭代中，一些配置属性被弃用并替换为新的、更具描述性的名称。
在原始配置中，同时使用了database.include.list（虽然可能配置不当）和database.whitelist、table.whitelist。这种混用可能导致配置冲突或预期外的行为，因为弃用属性可能不再被识别或处理。
结合上述分析，以下是一个优化后的Debezium MySQL连接器配置示例，假设我们要监控my_database中的user_table：
import io.debezium.config.Configuration;
// 假设Connection类包含getMysqlConnection()方法，返回MysqlConnection对象
// MysqlConnection对象包含getHost(), getPort(), getUsername(), getPassword(), getDbName(), getTables()等方法
public class DebeziumConfigLoader {
public static Configuration load(Connection connection) {
final MysqlConnection mysql = connection.getMysqlConnection();
return Configuration.create()
.with("name", "customer-mysql-connector")
.with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
.with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
.with("offset.flush.interval.ms", "60000")
.with("database.hostname", mysql.getHost())
.with("database.port", mysql.getPort())
.with("database.user", mysql.getUsername())
.with("database.password", mysql.getPassword())
// 假设mysql.getDbName()返回 "my_database"
// 明确指定要监控的数据库，例如 "my_database"
.with("database.include.list", mysql.getDbName())
// 如果需要监控多个数据库，例如 "db1,db2"
// .with("database.include.list", "my_database,another_database")
// 明确指定要监控的表，格式为 "数据库名.表名"
// 假设mysql.getTables()返回 ["user_table", "product_table"]
// 则应构造为 "my_database.user_table,my_database.product_table"
.with("table.include.list",
String.join(",", mysql.getTables().stream()
.map(tableName -> mysql.getDbName() + "." + tableName)
.toArray(String[]::new)))
.with("include.schema.changes", "false")
.with("database.server.id", "10181") // 确保此ID唯一
.with("database.server.name", "customer-mysql-db-server")
.with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
.with("database.history.file.filename", "/tmp/dbhistory.dat")
.with("offset.storage.file.filename", "/tmp/offsets.dat")
// 移除已弃用的 database.whitelist 和 table.whitelist
// 如果需要排除某些数据库或表，请使用 database.exclude.list 或 table.exclude.list
.build();
}
}在上述优化后的配置中，database.include.list被设置为实际的数据库名（例如my_database），而table.include.list则通过拼接数据库名.表名的方式，准确指定了需要监控的表。同时，移除了已弃用的whitelist属性。
配置完成后，通过线程池或其他方式提交Debezium引擎实例即可启动数据捕获：
executorService.submit(engine);
当Debezium捕获到数据变更事件时，notifying(this::handleEvent)中指定的handleEvent方法将被调用。在这个方法中，你可以解析ChangeEvent对象，获取变更类型（INSERT, UPDATE, DELETE）和变更前后的数据。
// 示例事件处理器
private void handleEvent(SourceRecord record) {
// 解析SourceRecord以获取实际的ChangeEvent
// ...
// 根据事件类型（op字段）判断是插入、更新还是删除
// 例如：
// Struct source = (Struct) record.value().get("source");
// String op = (String) record.value().get("op");
// if ("d".equals(op)) { // delete operation
// System.out.println("Data deleted: " + record);
// // 处理删除事件逻辑
// } else if ("c".equals(op)) { // create/insert operation
// // 处理插入事件
// } else if ("u".equals(op)) { // update operation
// // 处理更新事件
// }
System.out.println("Debezium captured event: " + record);
}SHOW VARIABLES LIKE 'log_bin'; -- 检查是否开启binlog SHOW VARIABLES LIKE 'binlog_format'; -- 检查binlog格式是否为ROW
如果未开启或格式不正确，需要在my.cnf中进行配置并重启MySQL：
[mysqld] log_bin = mysql-bin binlog_format = ROW server_id = 1 # 确保此ID在MySQL复制拓扑中唯一
Debezium在Java应用中捕获MySQL数据变更，特别是删除事件失败的问题，通常可归结为连接器配置不当。通过本文的详细解析和优化示例，我们强调了正确使用database.include.list和table.include.list的重要性，以及替换已弃用whitelist属性的必要性。遵循这些指导原则，并结合对MySQL Binlog配置和Debezium日志的细致检查，开发者可以有效地解决数据捕获问题，构建稳定可靠的CDC解决方案。
以上就是Debezium MySQL连接器数据变更捕获指南：解决配置陷阱的详细内容，更多请关注php中文网其它相关文章！
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移，垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是，许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号
191
收藏
