本文深入探讨了debezium在java应用中捕获mysql数据变更，特别是删除事件时常见的配置问题。核心内容包括正确配置`database.include.list`和`table.include.list`以精准指定监控范围，以及识别并替换已弃用的`database.whitelist`等属性。通过优化配置示例和注意事项，旨在帮助开发者构建稳定可靠的debezium cdc解决方案。
Debezium MySQL连接器核心配置与常见陷阱
Debezium作为一个强大的分布式变更数据捕获（CDC）平台，能够将数据库的每一次数据变更事件流式传输到消息队列中。然而，在实际应用中，尤其是在Java集成场景下，开发者可能会遇到Debezium无法捕获预期数据变更（例如删除操作）的问题。这通常源于MySQL连接器配置不当。本教程将详细解析Debezium MySQL连接器的关键配置，并指出常见的配置陷阱及其解决方案。
1. Debezium引擎与连接器概览
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连接器所需的配置。
2. 关键配置属性解析与优化
在配置Debezium MySQL连接器时，以下属性至关重要，并且是导致捕获失败的常见原因：
2.1 数据库与表的监控范围：include.list系列属性
Debezium提供了精细的控制来指定要监控的数据库和表。理解并正确使用这些属性是确保数据变更被捕获的关键。
-
database.include.list: 用于指定Debezium应该监控哪些数据库。多个数据库名之间使用逗号分隔。
- 错误示例: 将表名列表赋值给此属性，例如"database.include.list", "mydb.table1,mydb.table2"。这会导致Debezium尝试将mydb.table1识别为一个数据库，从而无法找到并监控实际的表。
- 正确用法: 如果只想监控my_database这个数据库，则应配置为"database.include.list", "my_database"。
-
table.include.list: 用于指定Debezium应该监控哪些表。表名必须是完全限定的，即数据库名.表名的格式。多个表之间使用逗号分隔。
- 正确用法: 如果要监控my_database中的user_table和product_table，则应配置为"table.include.list", "my_database.user_table,my_database.product_table"。
常见错误分析: 原始配置中可能存在将mysql.getTables()返回的表名列表直接赋给database.include.list的情况。这导致Debezium无法识别这些作为数据库名的表名，从而无法正确初始化监控。
2.2 弃用属性的替代方案
Debezium在不同版本迭代中，一些配置属性被弃用并替换为新的、更具描述性的名称。
-
database.whitelist 和 table.whitelist: 这些属性在较新的Debezium版本中已被弃用。
-
替代方案:
- database.whitelist 应替换为 database.include.list。
- table.whitelist 应替换为 table.include.list。
- 同样，database.blacklist 和 table.blacklist 也已被 database.exclude.list 和 table.exclude.list 替代。
-
替代方案:
在原始配置中，同时使用了database.include.list（虽然可能配置不当）和database.whitelist、table.whitelist。这种混用可能导致配置冲突或预期外的行为，因为弃用属性可能不再被识别或处理。
2.3 其他重要配置
- name: 连接器的逻辑名称，用于标识offset和历史记录。
- connector.class: 指定使用的连接器类，对于MySQL是io.debezium.connector.mysql.MySqlConnector。
- offset.storage: 偏移量存储机制。org.apache.kafka.connect.storage.FileOffsetBackingStore将偏移量存储到文件中，适用于单机测试或简单场景。生产环境通常使用Kafka Connect的分布式存储。
- database.hostname, database.port, database.user, database.password, database.dbname: MySQL数据库的连接参数。
- database.server.id: Debezium作为MySQL的客户端连接时使用的服务器ID，必须是唯一的且不同于MySQL复制拓扑中的任何其他服务器ID。
- database.server.name: Debezium连接器的逻辑服务器名称，用于构建Kafka主题名称。
- database.history: 数据库schema历史记录存储机制。io.debezium.relational.history.FileDatabaseHistory将历史记录存储到文件中。
- include.schema.changes: 是否将schema变更事件发送到Kafka。通常设置为false以只关注数据变更。
3. 优化后的配置示例
结合上述分析，以下是一个优化后的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属性。
4. 启动Debezium引擎与事件处理
配置完成后，通过线程池或其他方式提交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);
}5. 注意事项与最佳实践
-
MySQL Binlog配置: 确保MySQL服务器已启用Binlog，并且Binlog格式为ROW。这是Debezium捕获变更的基础。
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版本兼容性: 始终查阅您所使用的Debezium版本的官方文档，以了解最新的配置属性和弃用信息。属性名称和行为可能随版本更新而变化。
- 日志分析: 当遇到问题时，详细检查Debezium应用程序的日志以及MySQL服务器的错误日志。Debezium的日志通常会提供关于连接、配置解析和事件处理的有用信息。
- Offset和History文件: offset.storage.file.filename和database.history.file.filename指定的文件路径必须是应用程序可读写的。这些文件用于持久化Debezium的消费进度和数据库Schema历史，确保Debezium重启后能从上次停止的位置继续工作，并正确处理Schema变更。
- database.server.id唯一性: database.server.id对于Debezium连接器来说必须是唯一的。在MySQL复制集群中，它不能与任何其他MySQL服务器的server_id冲突。
总结
Debezium在Java应用中捕获MySQL数据变更，特别是删除事件失败的问题，通常可归结为连接器配置不当。通过本文的详细解析和优化示例，我们强调了正确使用database.include.list和table.include.list的重要性，以及替换已弃用whitelist属性的必要性。遵循这些指导原则，并结合对MySQL Binlog配置和Debezium日志的细致检查，开发者可以有效地解决数据捕获问题，构建稳定可靠的CDC解决方案。
