4、Debezium入门：架构、核心组件与支持数据库

好，咱们开始聊 Debezium。说实话，在 CDC 这个领域里，Debezium 是我个人最常用的工具之一。它基于 Kafka Connect 构建，天生就是为实时数据流而生的。你想想看，一个框架能帮你把数据库的变更事件一条不落地推送到消息队列里，这本身就是件很酷的事。

我记得第一次在生产环境用 Debezium 时，心里还有点打鼓——毕竟它要直接读数据库的 binlog，万一搞崩了怎么办？后来用熟了才发现，只要配置得当，它比想象中要稳得多。

4.1 Debezium 的整体架构

Debezium 的架构其实不复杂。说白了，它就是一个运行在 Kafka Connect 集群上的分布式 CDC 系统。每个数据库实例对应一个 Connector 任务，任务负责读取日志、解析事件、然后发给 Kafka。

我习惯把它的架构分成三层：

• 连接层（Connector）： 负责跟具体数据库打交道，比如 MySQL、PostgreSQL、MongoDB 等。
• 事件处理层： 包括 Snapshot（快照）和 Streaming（流式读取）两个阶段。
• 输出层： 把解析好的 Change Event 写入 Kafka Topic，或者通过 Sink Connector 写到其他系统。

下面这张图是我自己画的，能帮你快速理解 Debezium 的整体流程：

4.2 核心组件详解

4.2.1 Connector（连接器）

Connector 是 Debezium 的入口。每个 Connector 对应一个数据库实例，你只需要在 Kafka Connect 的 REST API 里提交一个 JSON 配置，它就会自动启动任务。

我个人觉得，Connector 的设计很巧妙——它把「连接数据库」和「解析事件」这两件事封装在一起，你基本不用写代码。配置好连接信息、表名、偏移量存储策略，剩下的交给它就行。

4.2.2 Snapshot（快照阶段）

Snapshot 是 Debezium 启动后的第一个动作。它会读取数据库当前的所有数据，为每张表生成一条条初始事件，写入 Kafka。

为什么会需要这个阶段？因为如果你只读 binlog，那 binlog 里只有启动之后的变更，之前的数据就丢了。Snapshot 就是为了补上这个缺口。

我曾经遇到过一个坑：某张表有 2 亿行数据，Snapshot 跑了整整 6 个小时还没跑完，导致下游系统一直收不到增量数据。后来我调整了 snapshot.fetch.size 和 max.queue.size，分批读取，才把时间压到 40 分钟。

4.2.3 Streaming（流式阶段）

Snapshot 完成后，Debezium 自动切换到 Streaming 模式。它会持续监听数据库的变更日志（比如 MySQL 的 binlog、PostgreSQL 的 WAL），每检测到一条变更，就立即生成一条 Change Event 发到 Kafka。

Streaming 的核心是「断点续传」。Debezium 会定期把当前读取到的 binlog 位置（Offset）保存到 Kafka 的一个特殊 Topic 里。如果进程挂了，重启后它会从上次记录的位置继续读，一条都不会丢。

嗯，这里要注意：Offset 的存储方式有两种——Kafka Topic 和文件系统。我习惯用 Kafka Topic，因为集群模式下多个 Worker 可以共享偏移量，不会冲突。

4.3 支持的数据库

Debezium 目前官方支持的数据库有这些：

数据库	Connector 名称	变更日志来源	个人评价
MySQL	io.debezium.connector.mysql.MySqlConnector	binlog（ROW 格式）	最成熟，我用得最多
PostgreSQL	io.debezium.connector.postgresql.PostgresConnector	WAL（逻辑复制槽）	性能好，但要注意复制槽管理
MongoDB	io.debezium.connector.mongodb.MongoDbConnector	oplog / Change Streams	适合文档型数据库同步
Oracle	io.debezium.connector.oracle.OracleConnector	LogMiner / XStream	配置稍复杂，需要 DBA 权限
SQL Server	io.debezium.connector.sqlserver.SqlServerConnector	CDC 表 + 变更跟踪	需要先开启数据库 CDC 功能
Db2	io.debezium.connector.db2.Db2Connector	IBM Db2 CDC 表	企业级场景用得较多
Cassandra	io.debezium.connector.cassandra.CassandraConnector	CommitLog	实验性，慎用于生产

从这张表你能看出来，Debezium 几乎覆盖了主流的关系型数据库和部分 NoSQL。我个人最推荐从 MySQL 或 PostgreSQL 入手，因为社区活跃、文档全、踩坑经验也多。

4.4 一个简单的配置示例

光说不练假把式。我给你看一个 MySQL Connector 的配置，这是我项目里常用的模板：

{
  "name": "mysql-connector-orders",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "debezium_pass",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "orders_db",
    "table.include.list": "orders_db.orders,orders_db.order_items",
    "snapshot.mode": "when_needed",
    "offset.storage": "org.apache.kafka.connect.storage.KafkaOffsetBackingStore",
    "offset.storage.topic": "debezium-offsets",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

配置好之后，用 curl 提交给 Kafka Connect 的 REST API：

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

几秒钟后，你就能在 Kafka 里看到 dbserver1.orders_db.orders 这样的 Topic 了。每条消息都对应数据库里的一行变更，结构清晰，直接消费就行。