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

好，咱们进入Debezium的世界。说实话，我第一次接触CDC时，面对一堆日志解析工具也挺头疼的。直到用了Debezium，才觉得“嗯，这才对嘛”。它把复杂的数据库日志读取，封装成了你熟悉的Kafka消息。你想想看，这不就是咱们数据工程师最舒服的姿势吗？

3.1 Debezium是什么？

Debezium是一个开源的分布式平台，专门用来捕获数据库的变更数据。它基于Kafka Connect构建，把数据库的INSERT、UPDATE、DELETE操作，实时转换成事件流，推送到Kafka里。

说白了，它就是数据库和消息队列之间的“翻译官”。你不需要懂MySQL的binlog格式，也不用管PostgreSQL的WAL日志怎么解析，Debezium全帮你搞定了。

3.2 Debezium架构：一张图看懂

我习惯先看架构图，再聊细节。下面这张图，是我自己总结的Debezium核心工作流程，你仔细看一遍，后面就好理解了。

数据流其实很简单：源数据库产生变更日志 → Debezium Connector 读取并解析 → 通过 Kafka Connect 发送到 Kafka Topic → 下游消费者订阅消费。每个步骤都解耦，这也是它稳定的原因。

3.3 核心组件详解

Debezium 的组件不多，但每个都很关键。我一个个说。

3.3.1 Connector（连接器）

这是Debezium的心脏。每个数据库类型对应一个Connector。比如MySQL有io.debezium.connector.mysql.MySqlConnector，PostgreSQL有PostgresConnector。Connector负责连接数据库、读取日志、解析事件。

3.3.2 Kafka Connect

Debezium是Kafka Connect的一个Source Connector实现。Kafka Connect负责管理Connector的生命周期：启动、停止、重启、配置更新。它还提供了REST API，你可以通过HTTP来管理Connector。

3.3.3 内部Topic

Debezium会创建几个内部Topic来管理状态：

• offset storage topic：记录当前读取到的binlog位置，用于断点续传
• schema history topic：记录数据库表结构的变更历史
• status topic：记录Connector的运行状态

这些Topic千万别手动删除，否则Debezium会丢失状态，重新全量读取数据。

3.4 支持的数据源

Debezium目前支持的主流数据库如下：

数据库	Connector 名称	依赖的日志	注意事项
MySQL	MySqlConnector	binlog (ROW格式)	需要开启 binlog，且格式为 ROW
PostgreSQL	PostgresConnector	WAL (逻辑复制槽)	需要设置 wal_level = logical
MongoDB	MongoDbConnector	oplog / Change Streams	副本集或分片集群
Oracle	OracleConnector	LogMiner / XStream	需要额外配置，较复杂
SQL Server	SqlServerConnector	CDC 表	需要启用数据库的 CDC 功能
Db2	Db2Connector	IBM Db2 CDC	企业级场景较多

3.5 环境搭建与配置

好，理论说完了，咱们动手。我以最常见的MySQL + Debezium为例，带你走一遍完整的搭建流程。

3.5.1 前提条件

• 已安装并运行Kafka和Zookeeper（版本建议2.8+）
• 已安装MySQL 5.7+ 或 8.0+
• MySQL已开启binlog，格式为ROW

3.5.2 下载与安装

Debezium通常以Kafka Connect插件的形式部署。你只需要下载Debezium的Connector插件包，放到Kafka Connect的插件目录下。

# 下载 Debezium MySQL Connector 插件
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.5.0.Final/debezium-connector-mysql-2.5.0.Final-plugin.tar.gz

# 解压到 Kafka Connect 插件目录
tar -xzf debezium-connector-mysql-2.5.0.Final-plugin.tar.gz -C /opt/kafka/plugins/

# 配置 Kafka Connect 的 worker 配置文件
# 编辑 config/connect-distributed.properties
echo "plugin.path=/opt/kafka/plugins" >> config/connect-distributed.properties

3.5.3 启动Kafka Connect

# 启动分布式模式的 Kafka Connect
bin/connect-distributed.sh config/connect-distributed.properties

启动后，你可以通过REST API检查是否正常运行：

curl -s http://localhost:8083/ | jq .

如果返回类似 {"version":"3.5.0","commit":"..."}，说明启动成功。

3.5.4 注册Debezium Connector

接下来，通过POST请求注册一个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_pass",
    "database.server.id": "184054",
    "database.server.name": "dbserver1",
    "database.include.list": "inventory",
    "table.include.list": "inventory.products,inventory.orders",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes.inventory",
    "include.schema.changes": "true",
    "snapshot.mode": "initial"
  }
}

然后执行：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ \
  -d @register-mysql.json

3.5.5 验证数据流

注册成功后，你可以用Kafka自带的控制台消费者，查看Topic中的数据：

# 查看所有Topic
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# 消费 dbserver1.inventory.products 这个Topic
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic dbserver1.inventory.products \
  --from-beginning

然后在MySQL中执行一条INSERT语句：

INSERT INTO inventory.products (name, price) VALUES ('新商品', 99.99);

你会立刻在消费者终端看到一条JSON消息，包含before（旧值，INSERT时为null）和after（新值）字段。嗯，这就是CDC的魅力——实时、准确、无侵入。

3.6 常见问题与避坑

• Connector启动失败：检查MySQL用户是否有REPLICATION SLAVE和REPLICATION CLIENT权限
• Topic没有数据：确认binlog格式为ROW，且表结构没有变化导致schema history不一致
• 性能问题：如果表数据量巨大，建议先做全量快照，再开启增量模式