3. Debezium入门：架构、核心组件与部署实战

Debezium 这个名字，你可能会觉得陌生。但说白了，它就是 CDC 领域的事实标准。我最早接触它是在 2018 年，那时候公司要做跨机房数据同步，调研了一圈，最后选了 Debezium。嗯，一用就是好几年。

这一章，我们不讲虚的。直接拆解 Debezium 的架构、核心组件，再给出一份能跑起来的配置。你跟着走一遍，就能在自己的环境里把数据实时抓上来。

3.1 Debezium 整体架构

先看一张图，把 Debezium 的骨架搭起来。

Debezium 核心架构 MySQL / PostgreSQL MongoDB / Oracle SQL Server / 其他 Source Connector Debezium Server Embedded Engine Converter（JSON / Avro） Kafka Topic Offset 存储 架构要点： • 数据源通过 Connector 接入，Debezium 支持三种部署模式 • Converter 负责将 Change Event 序列化为指定格式（JSON/Avro） • Offset 记录消费位置，保证重启后不丢数据、不重复消费

这张图里，我标出了三个关键层次：数据源、Connector 层、以及下游的 Kafka 和 Offset 存储。你仔细看，其实 Debezium 的核心逻辑很简单——监听数据库的 binlog 或 oplog，转成事件流，推给 Kafka。

3.2 核心组件拆解

3.2.1 Connector：数据接入的入口

Connector 是 Debezium 的「耳朵」。它负责跟数据库建立连接，读取变更日志。每种数据库都有自己的 Connector，比如：

  • MySQL Connector：基于 binlog，支持 GTID 和 ROW 格式
  • PostgreSQL Connector：基于逻辑复制槽（WAL）
  • MongoDB Connector：基于 oplog 或 Change Streams
  • SQL Server Connector：基于 CDC 表或变更跟踪
我的经验：MySQL Connector 是最成熟的，坑也最少。如果你刚开始接触 Debezium，建议先用 MySQL 练手。PostgreSQL 的逻辑复制槽有个问题——如果 Connector 挂了，复制槽不清理，会导致 WAL 堆积，把磁盘撑爆。我踩过这个坑，后来加了监控和自动清理脚本才解决。

3.2.2 Converter：数据格式的翻译官

Converter 负责把 Debezium 内部的数据结构转成你需要的格式。常用的有两种：

Converter 类型 序列化格式 适用场景 优缺点
JSONConverter JSON String 开发调试、简单消费 可读性好，但 Schema 信息丢失
AvroConverter Avro + Schema Registry 生产环境、跨团队协作 Schema 强校验，压缩率高，但依赖 Schema Registry

我个人习惯，开发阶段用 JSONConverter，方便看数据。上了生产，必须换成 AvroConverter。为什么？因为 Avro 带 Schema，下游消费者能明确知道字段类型，不会出现「这个字段是 String 还是 Long」的扯皮。

3.2.3 Offset：断点续传的命根子

Offset 是 Debezium 的「记忆」。它记录了当前已经读到哪个 binlog 位置了。如果进程重启，Debezium 会从 Offset 记录的位置继续读，保证不丢数据。

Offset 可以存在哪里？

  • 文件系统：默认存在 offset.dat 文件里，适合单机测试
  • Kafka Topic：用 offset.storage.kafka 配置，适合生产集群
  • 数据库：支持 JDBC 存储，适合已有数据库基础设施的团队
注意：我曾经遇到过一个线上事故——Offset 文件所在的磁盘满了，Debezium 写不进去 Offset，结果重启后重复消费了十几分钟的数据。后来我把 Offset 存储改到了 Kafka Topic，再也没出过类似问题。生产环境，千万别用文件存储 Offset。

3.3 三种部署模式

Debezium 支持三种部署方式，我分别说说它们的适用场景。

3.3.1 Kafka Connect 模式（最常用）

这是官方推荐的方式。Debezium 作为 Kafka Connect 的一个 Source Connector 运行。你只需要写一个 JSON 配置文件，通过 REST API 提交给 Kafka Connect 集群。

# 启动 Kafka Connect 时加载 Debezium 插件
connect-standalone.sh \
  connect-standalone.properties \
  mysql-connector.properties

优点：部署简单，支持分布式扩展，自带 REST API 管理。我大部分项目都用这个模式。

3.3.2 Debezium Server 模式

如果你不想用 Kafka，可以用 Debezium Server。它直接把 Change Event 推送到其他消息队列，比如 Pulsar、Kinesis、Google Pub/Sub。

# application.properties 示例
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname=192.168.1.100
debezium.source.database.port=3306
debezium.source.database.user=cdc_user
debezium.source.database.password=cdc_pass
debezium.sink.type=pulsar
debezium.sink.pulsar.broker.service.url=pulsar://localhost:6650

这个模式适合那些已经用了 Pulsar 或 Kinesis 的团队，不想再引入 Kafka。

3.3.3 Embedded Engine 模式

Embedded Engine 是把 Debezium 直接嵌入到你的 Java 应用里。适合做小规模、定制化的 CDC 场景。

// Java 代码示例
DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
    .using(props)
    .notifying(record -> {
        System.out.println("Key: " + record.key());
        System.out.println("Value: " + record.value());
    })
    .build();

engine.run();

嗯，这个模式我用的不多。除非是边缘场景，比如某个微服务需要监听自己的数据库变化做缓存刷新，否则不建议用 Embedded Engine。维护成本高，出了问题不好排查。

3.4 配置示例：从 MySQL 抓取数据到 Kafka

光说不练假把式。我们直接写一份能用的配置。

# mysql-connector.properties
name=mysql-connector-01
connector.class=io.debezium.connector.mysql.MySqlConnector
database.hostname=192.168.1.100
database.port=3306
database.user=cdc_user
database.password=cdc_pass
database.server.id=184054
database.server.name=myapp
database.include.list=mydb
table.include.list=mydb.orders,mydb.users
database.history.kafka.bootstrap.servers=kafka1:9092,kafka2:9092
database.history.kafka.topic=dbhistory.myapp
include.schema.changes=true
tombstones.on.delete=false
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
offset.storage.file.filename=/tmp/offsets.dat
offset.flush.interval.ms=10000
配置要点解读：
  • database.server.name：Kafka Topic 的前缀，比如 myapp.mydb.orders
  • table.include.list：只监听指定表，别全库抓，浪费资源
  • tombstones.on.delete=false：删除操作不生成墓碑消息，减少 Kafka 存储压力
  • offset.flush.interval.ms：Offset 刷盘间隔，太短影响性能，太长可能丢数据

配置写好后，通过 Kafka Connect REST API 提交：

curl -X POST -H "Content-Type: application/json" \
  --data-binary @mysql-connector.properties \
  http://localhost:8083/connectors

然后你就可以在 Kafka 里看到类似 myapp.mydb.orders 的 Topic 了。消费一下试试：

kafka-console-consumer.sh \
  --bootstrap-server kafka1:9092 \
  --topic myapp.mydb.orders \
  --from-beginning

你会看到类似这样的 JSON 数据：

{
  "before": null,
  "after": {
    "id": 1001,
    "user_id": 42,
    "amount": 299.00,
    "status": "paid"
  },
  "source": {
    "db": "mydb",
    "table": "orders",
    "ts_ms": 1700000000000
  },
  "op": "c"
}

op 字段表示操作类型：c 是创建，u 是更新，d 是删除。before 和 after 分别记录变更前后的数据。

一个小技巧：刚开始调试时，把 key.converter.schemas.enable 和 value.converter.schemas.enable 都设为 false，这样输出的 JSON 不带 Schema 信息，看起来清爽。等调试通了，再改成 Avro 上生产。

3.5 避坑指南

最后，分享几个我踩过的坑：

  • MySQL binlog 格式必须设为 ROW：如果用的是 STATEMENT 或 MIXED，Debezium 拿不到完整的数据变更。我曾经接手过一个项目，binlog 格式是 MIXED，结果更新操作只记录了 SQL，没有字段值，根本没法用。
  • 数据库用户权限要够：至少需要 SELECT、RELOAD、REPLICATION SLAVE、REPLICATION CLIENT 权限。少了任何一个，Connector 启动就会报错。
  • Kafka Topic 分区数要提前规划：Debezium 默认每个表一个 Topic，分区数默认 1。如果你的表数据量大，建议提前设好分区数，比如 6 或 12。Topic 创建后再改分区数很麻烦。
  • Offset 存储别用文件：前面说过了，生产环境一定要用 Kafka Topic 或数据库存储 Offset。

好了，这一章的内容就到这里。Debezium 的架构和配置其实不复杂，核心就是 Connector 抓数据、Converter 转格式、Offset 记位置。你照着上面的配置跑一遍，应该半小时内就能看到数据流进 Kafka。

← 上一章 📖 返回目录 下一章 →