📚 课程目录第4章 Debezium深度实践（上）

4、Debezium深度实践（上）：架构、核心概念与嵌入式模式

好，咱们进入 Debezium 的实战环节。说实话，CDC 工具不少，Canal、Maxwell、DataBus 我都用过，但 Debezium 是我个人最推荐的一个。为什么？因为它跟 Kafka 生态结合得太紧密了，而且社区活跃，文档也全。

这一章我们先打基础，把 Debezium 的架构、核心概念讲透，再上手配置 MySQL Binlog 监听，最后聊聊嵌入式模式——这个模式很多人不知道，但其实特别适合轻量级场景。

4.1 Debezium 架构总览

先看整体架构。Debezium 本质上是一个 CDC 连接器框架，它基于 Kafka Connect 构建。你想想看，它做的事情很简单：监听数据库的变更日志，然后把变更事件转换成结构化的消息，推送到 Kafka。

核心流程：

数据库产生变更（INSERT/UPDATE/DELETE）
Debezium Connector 读取 Binlog（或 WAL）
生成 Change Event，写入 Kafka Topic
下游消费者消费这些事件

我画了一张架构图，你看一眼就明白了：

4.2 三大核心概念：Connector、Snapshot、Offset

这三个概念你搞懂了，Debezium 就掌握了一半。

4.2.1 Connector（连接器）

Connector 是 Debezium 的“心脏”。每个数据库类型对应一个 Connector，比如 MySqlConnector、PostgresConnector、MongoDbConnector。它的职责就两件事：

配置管理：告诉 Debezium 连哪个数据库、监听哪些表、用什么格式输出
生命周期管理：启动 Snapshot、切换到 Streaming、处理故障恢复

我记得第一次配置 MySQL Connector 时，漏了一个 database.include.list 参数，结果把整个实例的表都监听了，Kafka 里瞬间多了几百个 Topic……嗯，教训深刻。

配置示例：

{
  "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_pwd",
    "database.server.id": "223344",
    "database.server.name": "prod_orders",
    "database.include.list": "order_db",
    "table.include.list": "order_db.orders,order_db.order_items",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.orders",
    "include.schema.changes": "true"
  }
}

4.2.2 Snapshot（快照）

Snapshot 是 Debezium 启动时的“第一件事”。它会把当前数据库中的存量数据全部读出来，写入 Kafka。为什么要做这个？因为 Binlog 只记录增量变更，存量数据你得先捞一遍。

Snapshot 有三种模式：

模式	说明	适用场景
initial	启动时执行完整快照，完成后切到流模式	默认模式，首次部署
when_needed	只在需要时触发快照（比如表结构变更后）	已有快照，增量维护
never	跳过快照，直接开始流读取	只关心增量数据

注意：Snapshot 期间会锁表吗？默认情况下，MySQL Connector 使用 snapshot.locking.mode 参数控制锁行为。我建议生产环境设置为 minimal，只锁结构不锁数据，避免影响业务。

4.2.3 Offset（偏移量）

Offset 是 Debezium 的“记忆”。它记录了当前已经读到 Binlog 的哪个位置。一旦进程重启，Debezium 会从 Offset 记录的位置继续读取，不会重复也不会丢失。

Offset 存储在哪里？默认存在 Kafka 的 connect-offsets Topic 里。你也可以配置成文件存储或数据库存储。

我曾经遇到过一个坑：Kafka 集群故障导致 Offset Topic 数据丢失，重启后 Debezium 从最开始的 Binlog 位置重新读取，结果把过去三个月的数据又发了一遍……下游数仓直接炸了。所以，Offset 的持久化一定要做好，建议开启 Kafka 的日志压缩和副本机制。

4.3 MySQL Binlog 监听配置实战

好，理论讲完了，咱们动手配一个 MySQL 的 Binlog 监听。我会把每一步拆开讲，你跟着做就行。

4.3.1 MySQL 端准备

首先，MySQL 必须开启 Binlog。检查一下你的 MySQL 配置文件：

[mysqld]
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 7

几个关键点：

binlog_format 必须为 ROW：只有行模式才能记录每一行的变更细节
binlog_row_image 建议 FULL：这样 Debezium 能拿到变更前后的完整数据
expire_logs_days：根据你的磁盘空间设置，太短会导致 Offset 过期无法回溯

然后创建一个 Debezium 专用用户：

CREATE USER 'debezium'@'%' IDENTIFIED BY 'debezium_pwd';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

注意，REPLICATION SLAVE 和 REPLICATION CLIENT 是必须的，否则 Debezium 读不了 Binlog。

4.3.2 Kafka Connect 配置

Debezium 通常以 Kafka Connect 插件的形式运行。把 Debezium 的 JAR 包放到 Kafka Connect 的插件目录，然后启动 Connect：

# config/connect-distributed.properties
bootstrap.servers=localhost:9092
group.id=connect-cluster
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.topic=connect-offsets
config.storage.topic=connect-configs
status.storage.topic=connect-status
plugin.path=/kafka/plugins

启动后，通过 REST API 提交 Connector 配置：

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "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_pwd",
      "database.server.id": "223344",
      "database.server.name": "prod_orders",
      "database.include.list": "order_db",
      "table.include.list": "order_db.orders",
      "database.history.kafka.bootstrap.servers": "kafka:9092",
      "database.history.kafka.topic": "schema-changes.orders",
      "include.schema.changes": "true"
    }
  }'

配置成功后，你会在 Kafka 里看到 prod_orders.order_db.orders 这个 Topic，里面就是实时变更事件。

4.4 Debezium Engine 嵌入式模式

最后聊聊嵌入式模式。这个模式很多人不知道，但我觉得特别实用。

传统模式下，Debezium 依赖 Kafka Connect 和 Kafka 集群。但有些场景你不想引入 Kafka，比如：

只想把数据库变更直接推送到下游应用
测试环境不想搭一套 Kafka
轻量级的数据同步任务

这时候，Debezium Engine 就派上用场了。它把 Debezium 的核心能力封装成一个 Java 库，你可以在自己的应用里直接启动一个 CDC 引擎，不需要 Kafka。

4.4.1 嵌入式模式架构

说白了，嵌入式模式就是把 Connector、Snapshot、Offset 管理全部内嵌到你的 JVM 进程里。你只需要提供一个 Consumer 接口，Debezium 会把变更事件直接推给你。

嵌入式模式 vs 标准模式：

对比项	标准模式（Kafka Connect）	嵌入式模式（Engine）
依赖	Kafka + Kafka Connect	仅 Debezium 库
部署	独立集群	内嵌到应用
扩展性	高，支持多 Connector	低，适合单实例
适用场景	生产级数据管道	轻量级同步、测试

4.4.2 代码实战

来看一个完整的嵌入式模式示例。我用的是 Debezium 1.9+ 版本，Maven 依赖如下：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-api</artifactId>
    <version>1.9.8.Final</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>1.9.8.Final</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-mysql</artifactId>
    <version>1.9.8.Final</version>
</dependency>

然后写一个主类：

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class EmbeddedEngineDemo {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.setProperty("name", "engine-mysql-orders");
        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", "/tmp/offsets.dat");
        props.setProperty("offset.flush.interval.ms", "60000");
        props.setProperty("database.hostname", "192.168.1.100");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", "debezium");
        props.setProperty("database.password", "debezium_pwd");
        props.setProperty("database.server.id", "223344");
        props.setProperty("database.server.name", "embedded_orders");
        props.setProperty("database.include.list", "order_db");
        props.setProperty("table.include.list", "order_db.orders");
        props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
        props.setProperty("database.history.file.filename", "/tmp/dbhistory.dat");

        try (DebeziumEngine<ChangeEvent<String, String>> engine =
                 DebeziumEngine.create(Json.class)
                     .using(props)
                     .notifying(records -> {
                         for (ChangeEvent<String, String> r : records) {
                             System.out.println("Key: " + r.key());
                             System.out.println("Value: " + r.value());
                             System.out.println("---");
                         }
                     })
                     .build()) {

            ExecutorService executor = Executors.newSingleThreadExecutor();
            executor.execute(engine);

            // 让引擎运行 60 秒
            Thread.sleep(60000);
            engine.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

这段代码做了几件事：

配置 MySQL 连接信息和监听表
使用文件存储 Offset 和 Schema 历史
启动引擎，把变更事件打印到控制台

个人建议：嵌入式模式非常适合做“数据库变更 -> 实时通知”的场景。我之前在一个订单系统中，用嵌入式模式监听订单状态变更，直接推送到 WebSocket 给前端做实时展示，省掉了 Kafka 这一层，延迟在 50ms 以内。

注意：嵌入式模式是单进程的，如果进程挂了，CDC 就停了。生产环境建议配合进程守护工具（如 Supervisor）或容器编排（K8s）来保证高可用。

4.5 本章小结

这一章我们覆盖了 Debezium 的核心内容：

架构：Connector 内部有 Snapshotter、Streaming、Offset Manager、Event Converter 四大组件
三大概念：Connector 负责连接、Snapshot 负责存量、Offset 负责断点
MySQL 配置：Binlog 必须为 ROW 模式，用户需要 REPLICATION 权限
嵌入式模式：轻量级 CDC，适合不需要 Kafka 的场景