4. Debezium MySQL Connector 实战：配置MySQL源端、开启Binlog、全量+增量同步配置
好，咱们进入实战环节。前面讲了CDC的理论和Debezium的架构，说白了都是热身。这一章咱们真刀真枪地干——把MySQL源端配好，把Binlog打开，然后让Debezium把全量和增量数据都抓出来。
我个人习惯，做数据同步之前，先把源端的环境摸清楚。MySQL的Binlog如果不打开，后面所有工作都是白搭。你想想看，Debezium再牛，它也得有数据源啊。
4.1 MySQL源端配置：Binlog必须开
MySQL默认是不开Binlog的，尤其是云上RDS，很多默认配置都是关闭状态。我记得有一次帮客户做迁移，折腾了半天发现Debezium连不上，最后一看，Binlog没开……嗯，这种坑踩过一次就记住了。
先检查你的MySQL是否开启了Binlog：
-- 登录MySQL后执行
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';
SHOW VARIABLES LIKE 'binlog_row_image';如果 log_bin 是 OFF，那就要改配置了。找到MySQL的配置文件（通常是 /etc/my.cnf 或 /etc/mysql/mysql.conf.d/mysqld.cnf），加上以下内容：
[mysqld]
# 必须开启
log_bin = mysql-bin
# 必须用ROW模式
binlog_format = ROW
# 建议用FULL，记录完整行数据
binlog_row_image = FULL
# 建议设置，保留7天
expire_logs_days = 7
# 每个Binlog文件大小
max_binlog_size = 1G
# 要同步的数据库（可选，但推荐）
binlog_do_db = your_database_name重启后再次验证：
SHOW MASTER STATUS;看到类似下面的输出，就说明Binlog已经正常工作了：
+------------------+----------+--------------+------------------+
| File | Position | Binlog_Do_DB | Binlog_Ignore_DB |
+------------------+----------+--------------+------------------+
| mysql-bin.000003 | 154 | your_db | |
+------------------+----------+--------------+------------------+4.2 创建Debezium专用的MySQL用户
我不建议直接用root用户。安全是一方面，另一方面是权限控制要清晰。Debezium需要哪些权限？我列个清单：
| 权限 | 用途 |
|---|---|
SELECT |
读取全量数据 |
RELOAD |
执行 FLUSH TABLES WITH READ LOCK（全量快照时用） |
REPLICATION SLAVE |
读取Binlog |
REPLICATION CLIENT |
获取Binlog位置信息 |
SHOW DATABASES |
列举数据库（可选） |
创建用户的SQL如下：
CREATE USER 'debezium'@'%' IDENTIFIED BY 'your_strong_password';
GRANT SELECT, RELOAD, REPLICATION SLAVE, REPLICATION CLIENT, SHOW DATABASES
ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;*.* 换成 your_database_name.*，权限更小，更安全。
4.3 配置Debezium MySQL Connector
好，源端准备好了，咱们开始配Connector。Debezium的配置是通过REST API提交给Kafka Connect的。我个人习惯先把配置写成JSON文件，方便管理和版本控制。
下面是一个完整的配置示例：
{
"name": "mysql-connector-demo",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "192.168.1.100",
"database.port": "3306",
"database.user": "debezium",
"database.password": "your_strong_password",
"database.server.id": "12345",
"database.server.name": "mysql-server-01",
"database.include.list": "your_database_name",
"table.include.list": "your_database_name.users,your_database_name.orders",
"database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
"database.history.kafka.topic": "schema-changes.mysql-server-01",
"include.schema.changes": "true",
"snapshot.mode": "initial",
"tombstones.on.delete": "false",
"decimal.handling.mode": "string",
"time.precision.mode": "connect"
}
}这里有几个关键参数，我重点说一下：
- database.server.id：每个Connector实例必须唯一，不能和MySQL集群中其他server-id冲突。
- database.server.name：这个值会作为Kafka topic的前缀，比如
mysql-server-01.your_database_name.users。 - snapshot.mode：
initial表示先做全量快照，然后自动切换到增量模式。这是最常用的模式。 - tombstones.on.delete：建议设为
false，否则删除操作会生成墓碑消息，增加存储开销。
4.4 全量+增量同步的核心逻辑
Debezium的同步流程，说白了就两步：先拍快照，再追日志。我画了一张图，帮你理解这个过程：
这个流程里有个关键点：全量快照开始时，Debezium会先记录当前的Binlog位置（文件名+偏移量）。然后才开始读全量数据。等全量数据读完了，它就从刚才记录的位置开始消费Binlog。这样一来，全量快照期间产生的增量变更也不会丢。
4.5 提交Connector并验证
配置写好了，怎么提交？用Kafka Connect的REST API：
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ \
-d @mysql-connector-config.json提交成功后，检查状态：
curl http://localhost:8083/connectors/mysql-connector-demo/status如果看到 "state": "RUNNING"，说明Connector已经正常工作了。这时候去Kafka里看看topic：
kafka-topics.sh --bootstrap-server kafka1:9092 --list应该能看到类似 mysql-server-01.your_database_name.users 这样的topic。
kafka-console-consumer.sh --bootstrap-server kafka1:9092 \
--topic mysql-server-01.your_database_name.users \
--from-beginning4.6 避坑指南
我在项目中遇到过几个常见问题，这里列出来，你遇到了可以直接对照排查：
- Connector启动失败，报权限错误：检查MySQL用户权限，尤其是
REPLICATION SLAVE和RELOAD是否授予。 - 全量快照超时：如果表数据量很大（比如上亿行），默认的快照超时时间可能不够。可以在配置里加
"snapshot.fetch.size": "10000"来分批读取。 - Binlog被清理导致增量中断：我曾经遇到过，全量快照跑了太久，Binlog被自动清理了。解决办法是适当调大
expire_logs_days，或者先做一次全量备份再启动Connector。 - Decimal类型精度丢失：MySQL的Decimal类型在Debezium默认会转成double，有精度问题。建议配置
"decimal.handling.mode": "string"，用字符串保留原始精度。