第一个Connector：MySQL Connector
好，咱们直接动手。前面两章把Debezium的架构和部署聊完了，这一章我们真刀真枪地跑一个MySQL Connector。说白了，就是让Debezium去读MySQL的binlog，然后把数据变化实时吐出来。
我个人习惯，学一个新工具，先让它跑起来，再回头琢磨细节。所以这一章，我们重点解决三件事：怎么配、怎么启动、怎么看数据流。最后再聊聊我踩过的那些坑。
配置MySQL Connector
配置Connector，本质上就是告诉Debezium三件事：连哪个MySQL、监听哪些表、数据往哪发。我用的是Kafka Connect的REST API方式提交配置，这也是生产环境最常用的方式。
先看一个最简配置，我把它存成 mysql-connector.json：
{
"name": "inventory-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "192.168.1.100",
"database.port": "3306",
"database.user": "debezium",
"database.password": "debezium@123",
"database.server.id": "223344",
"database.server.name": "dbserver1",
"database.include.list": "inventory",
"table.include.list": "inventory.products,inventory.orders",
"database.history.kafka.bootstrap.servers": "kafka1:9092,kafka2:9092",
"database.history.kafka.topic": "schema-changes.inventory"
}
}这里有几个关键参数，我拎出来说一下：
| 参数 | 说明 | 我的建议 |
|---|---|---|
database.server.id |
Debezium伪装成一个MySQL Slave，这个ID不能和集群里其他Slave重复 | 用时间戳后四位 + 随机数，基本不会撞 |
database.server.name |
Kafka Topic的前缀，所有数据流都基于这个名字 | 用业务域名或项目代号，别用test123这种 |
database.include.list |
要监听的数据库，多个用逗号分隔 | 尽量精确，别偷懒写* |
table.include.list |
要监听的表，格式是库名.表名 | 只监听需要的表，减少不必要的binlog消费 |
database.history.kafka.topic |
存储DDL历史，Debezium重启时靠它恢复Schema | 一个Connector一个Topic，别混用 |
database.server.name一旦确定，不要随意更改。我见过有人改了这个参数，结果Kafka Topic全乱了，历史数据对不上，最后只能重新全量同步。嗯，血的教训。
启动任务
配置写好了，怎么启动？用Kafka Connect的REST API，一行命令搞定：
curl -i -X POST -H "Accept:application/json" \
-H "Content-Type:application/json" \
http://localhost:8083/connectors/ \
-d @mysql-connector.json返回HTTP 201，说明Connector注册成功。这时候你可以查一下状态：
curl http://localhost:8083/connectors/inventory-connector/status返回结果大概长这样：
{
"name": "inventory-connector",
"connector": {
"state": "RUNNING",
"worker_id": "192.168.1.10:8083"
},
"tasks": [
{
"id": 0,
"state": "RUNNING"
}
]
}看到 state: "RUNNING"，说明任务跑起来了。如果状态是 FAILED 或 PAUSED，别慌，后面我会讲怎么排查。
tail -f connect.log | grep -i error 实时监控。很多问题在日志里会提前暴露，不用等状态变成FAILED才去查。
查看数据流
任务跑起来了，数据流在哪看？Debezium会把变更事件写到Kafka Topic里。Topic的命名规则是：{database.server.name}.{database}.{table}。
拿我们刚才的配置举例：
dbserver1.inventory.products— products表的数据变更dbserver1.inventory.orders— orders表的数据变更dbserver1— 这是元数据Topic，存Schema变更历史
用Kafka自带的命令行工具，直接消费看看：
kafka-console-consumer \
--bootstrap-server kafka1:9092 \
--topic dbserver1.inventory.products \
--from-beginning \
--property print.key=true \
--property print.value=true这时候去MySQL里执行一条INSERT：
INSERT INTO inventory.products (name, price) VALUES ('机械键盘', 399.00);你会在消费者终端看到类似这样的输出：
{
"schema": { ... },
"payload": {
"before": null,
"after": {
"id": 105,
"name": "机械键盘",
"price": 399.00
},
"source": {
"version": "1.9.7.Final",
"connector": "mysql",
"name": "dbserver1",
"ts_ms": 1695000000000,
"snapshot": "false",
"db": "inventory",
"table": "products",
"server_id": 223344,
"gtid": null,
"file": "mysql-bin.000023",
"pos": 45678,
"row": 0
},
"op": "c",
"ts_ms": 1695000000123
}
}这个JSON结构，我拆开说一下：
- before：变更前的数据。INSERT操作是null，UPDATE和DELETE会有值。
- after：变更后的数据。DELETE操作是null。
- source：元数据，记录了这条变更来自哪个库、哪个表、binlog的哪个位置。排查问题的时候，这个字段特别有用。
- op：操作类型。c=创建，u=更新，d=删除，r=读取（全量快照时出现）。
常见问题排查
跑起来容易，跑稳了难。我把自己在项目中遇到过的几个典型问题列出来，你遇到了可以直接对照排查。
问题1：Connector状态一直是FAILED
最常见的原因是MySQL用户权限不够。Debezium需要的权限有：SELECT、RELOAD、REPLICATION SLAVE、REPLICATION CLIENT。检查一下：
SHOW GRANTS FOR 'debezium'@'%';如果缺了 REPLICATION SLAVE，Debezium连不上binlog，直接报错。
问题2：数据没出来，但状态是RUNNING
这种情况最让人抓狂。我遇到过两次，一次是 table.include.list 写错了表名，一次是MySQL的binlog格式没设对。检查两件事：
- MySQL的binlog格式必须是
ROW：SHOW VARIABLES LIKE 'binlog_format'; - 确认表名大小写：MySQL在Linux上默认区分大小写，
Products和products是两张表。
问题3：重启后数据重复或丢失
这通常和 database.history.kafka.topic 有关。如果这个Topic被删了或者数据过期了，Debezium重启后无法恢复Schema，就会出问题。我的建议：
- 给这个Topic设置
log.retention.ms=-1，永不过期 - 或者设置一个足够长的保留期，比如30天
核心流程一览
为了让你对整个流程有个直观的印象，我画了一张图。从MySQL binlog到Kafka Topic，每一步都清晰可见。
这张图把整个链路串起来了。你从MySQL写入一条数据，经过Debezium解析，最终落到Kafka Topic里，消费者可以实时消费。每一步都有明确的职责，排查问题时，按图索骥就好。