3. Debezium入门：它到底是什么？
Debezium，说白了就是一个能「偷看」数据库变更日志的开源工具。我最早接触它是在2018年，当时公司要做实时数仓，每天全量同步MySQL数据，凌晨跑批跑到天亮。后来我发现了Debezium——嗯，这玩意儿确实改变了我的工作方式。
它的核心能力很简单：把数据库的INSERT、UPDATE、DELETE操作，实时转化成事件流。你想想看，以前我们要监听数据变更，要么写触发器，要么轮询查表，要么改业务代码埋点。这些方式要么侵入性强，要么延迟高，要么维护成本大。Debezium直接读取数据库的binlog（MySQL）、WAL（PostgreSQL）或oplog（MongoDB），不碰业务表，不写一行代码。
一句话总结：Debezium = 数据库的「日志窃听器」 + 「事件转换器」 + 「Kafka生产者」。
3.1 支持的数据源
Debezium目前支持三大主流数据库，我按使用频率排个序：
| 数据库 | 读取的日志 | 典型场景 | 我的经验 |
|---|---|---|---|
| MySQL | binlog（ROW格式） | 电商订单、用户中心 | 最成熟，坑最少 |
| PostgreSQL | WAL（逻辑复制槽） | 金融、物联网时序数据 | 需要提前创建复制槽 |
| MongoDB | oplog（副本集） | 文档型数据、日志存储 | 注意副本集必须启用 |
我个人习惯，新项目首选MySQL + Debezium。为什么？因为MySQL的binlog机制最成熟，社区文档最全，遇到问题基本都能搜到答案。PostgreSQL的WAL逻辑复制槽需要手动管理，我曾经在生产环境忘记清理复制槽，导致WAL文件堆积，磁盘爆了——嗯，那次教训挺深刻的。
避坑指南：我曾经在MongoDB副本集上踩过坑——如果副本集成员少于3个，oplog可能被自动清理，导致Debezium丢失变更事件。生产环境至少部署3个节点。
3.2 架构组件：三大核心模块
Debezium的架构不复杂，就三个核心组件。我画了一张图帮你理解它们的关系：
3.3 Connector：配置驱动的连接器
Connector是Debezium的「入口」。你只需要写一个JSON配置文件，告诉它：
- 连接哪个数据库（host、port、用户名、密码）
- 监听哪些表（table.include.list）
- 把事件发到哪个Kafka Topic（topic.prefix）
举个例子，一个MySQL Connector的配置长这样：
{
"name": "my-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "192.168.1.100",
"database.port": "3306",
"database.user": "debezium",
"database.password": "debezium_pwd",
"database.server.id": "184054",
"database.server.name": "myapp",
"database.include.list": "mydb",
"table.include.list": "mydb.orders,mydb.users",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "schema-changes.myapp"
}
}这里有个关键点——database.server.name。我刚开始用的时候，以为这只是个随便填的标识符。后来发现，它直接决定了Kafka Topic的前缀。比如你填了myapp，那么orders表的变更事件就会发到myapp.mydb.orders这个Topic。嗯，命名规范很重要，不然Topic一多就乱了。
小技巧：我个人习惯把database.server.name设为应用名+环境，比如order-prod、user-staging。这样在Kafka里一眼就能看出Topic属于哪个系统。
3.4 Snapshot：全量快照
第一次启动Connector时，Debezium会先做一次全量快照。为什么需要这一步？
你想想看，如果数据库里已经有100万条订单数据，而Debezium只从当前时刻开始监听，那之前的100万条就永远丢了。Snapshot的作用就是：先把存量数据全部读出来，发到Kafka，然后再切换到增量监听。
Snapshot的执行模式有四种：
| 模式 | 行为 | 适用场景 |
|---|---|---|
| initial | 先快照，后流式 | 默认模式，首次部署 |
| when_needed | 只在需要时触发 | 恢复丢失数据 |
| never | 跳过快照，直接流式 | 已有全量数据，只需增量 |
| schema_only | 只捕获表结构，不读数据 | 只关心未来变更 |
我在项目中遇到过一个问题：一张表有5000万行数据，Snapshot跑了3个小时还没跑完，Kafka的磁盘快满了。后来我改用schema_only模式，先只同步表结构，然后用一个离线脚本把全量数据批量导入Kafka，最后再切回流式模式。嗯，有时候「不按套路出牌」反而更高效。
注意：Snapshot期间，Debezium会锁表吗？默认情况下，MySQL Connector的Snapshot会加READ LOCK，防止数据不一致。如果你的表很大，建议在低峰期执行，或者使用snapshot.locking.mode=none（但要确保数据一致性）。
3.5 Streaming：增量流式捕获
Snapshot完成后，Debezium自动进入Streaming模式。这时候它就像个「永不休息的哨兵」，持续读取数据库的变更日志。
Streaming的核心机制：
- 轮询日志：每隔
poll.interval.ms（默认500ms）检查一次日志是否有新变更 - 解析事件：把binlog/WAL/oplog的二进制格式，解析成结构化的JSON事件
- 发送到Kafka：每个变更事件包含
before（旧值）和after（新值）两个字段
一个典型的变更事件长这样：
{
"schema": { ... },
"payload": {
"before": null,
"after": {
"id": 1001,
"name": "张三",
"status": "ACTIVE"
},
"source": {
"db": "mydb",
"table": "users",
"ts_ms": 1700000000000
},
"op": "c",
"ts_ms": 1700000001000
}
}注意op字段：c表示创建（Create）、u表示更新（Update）、d表示删除（Delete）、r表示快照读取（Read）。这个字段在后续做增量计算时非常有用——你可以根据操作类型决定怎么处理。
我曾经遇到一个坑：Streaming模式下，如果Kafka挂了，Debezium会不断重试。默认重试间隔是1秒，重试次数不限。结果有一次Kafka集群重启花了10分钟，Debezium在这10分钟里产生了大量重试日志，把磁盘打满了。后来我加了max.retries和retry.backoff.ms参数，限制重试次数和间隔。
我的建议：生产环境一定要配置offset.storage.file.filename（文件存储）或offset.storage.topic（Kafka存储）。这样即使Debezium重启，也能从上次中断的位置继续消费，不会重复也不会丢失。
3.6 小结：Debezium到底解决了什么问题？
回到最开始的问题——我们为什么要用Debezium？
- 无侵入：不碰业务代码，不改表结构
- 低延迟：毫秒级捕获变更，比轮询快得多
- 全量+增量：Snapshot搞定存量，Streaming搞定增量
- 生态好：输出到Kafka，下游随便接Flink、Spark、Elasticsearch