ZH
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish日本語한국어简体中文
首页 / 架构与设计 / Kafka

使用 Debezium 和 Kafka 实现变更数据捕获

Cris The Dev · 2026/10/12 · 8 分钟阅读 · 312

1. 概述

当一个服务需要知道另一个数据库中的数据刚刚发生了变化时，常见的做法是先写入数据库，然后再发送一条消息到 Kafka。这种方式称为双写，容易导致数据不一致。变更数据捕获（Change Data Capture，CDC）通过直接读取数据库的事务日志来解决这个问题。

在这篇文章中，你将使用 Docker Compose 构建 PostgreSQL、Kafka 和 Debezium，然后注册一个连接器。接下来你将编写一个 Spring Boot 消费者来处理 orders 表的 INSERT、UPDATE 和 DELETE 事件。

2. 为什么需要 CDC

让我们看一下下面这个熟悉的双写代码段：

@Transactional
public void placeOrder(Order order) {
    orderRepository.save(order);
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
}

这段代码有两个弱点：

  • 如果 Kafka 在 save 成功后出错，数据库有订单，但没有发送任何事件。
  • 如果 send 成功但事务回滚，消费者会收到一个不存在的订单的事件。

CDC 反转了这个流程。应用程序只写入数据库。一个单独的进程读取事务日志（PostgreSQL 使用 WAL，MySQL 使用 binlog），为每个已提交的更改发出事件。因为日志只包含已提交的更改，事件总是与实际数据匹配。

Debezium 是一套 CDC 连接器，运行在 Kafka Connect 上。数据流如下：

应用程序 ──写入──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafka topic ──> 消费者

3. 搭建环境

3.1. Docker Compose

你需要三个容器：PostgreSQL 启用逻辑复制、Kafka 运行在 KRaft 模式下以及 Kafka Connect 已预装 Debezium 插件。

services:
  postgres:
    image: postgres:16
    command: ["postgres", "-c", "wal_level=logical"]
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    ports:
      - "5432:5432"

  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: quay.io/debezium/connect:2.7
    depends_on: [kafka, postgres]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses

这个文件中有三个值得注意的地方：

  • wal_level=logical 是必需的。缺少这个参数，Debezium 无法创建复制槽。
  • Kafka 有两个监听器。容器使用 kafka:29092，而 Spring Boot 应用程序运行在你的机器上使用 localhost:9092。
  • Kafka Connect 将配置、偏移量和状态保存到三个单独的 topic。这样连接器重启后会从 WAL 中的旧位置继续读取。

启动并创建表：

docker compose up -d
docker compose exec postgres psql -U postgres -d shop -c "
  CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer_email VARCHAR(255) NOT NULL,
    status VARCHAR(32) NOT NULL,
    total NUMERIC(10,2) NOT NULL
  );
  ALTER TABLE orders REPLICA IDENTITY FULL;"

REPLICA IDENTITY FULL 要求 PostgreSQL 将整个旧行写入 WAL。如果没有这个命令，UPDATE 事件中的 before 字段将为 null，而 DELETE 事件只包含主键。

3.2. 注册连接器

Kafka Connect 在 8083 端口提供 REST API。你可以如下注册 PostgreSQL 连接器：

curl -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "shop_slot",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "string",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}'

重要属性的含义：

  • plugin.name: pgoutput 使用 PostgreSQL 10+ 内置的解码插件，所以你不需要安装额外的扩展。
  • topic.prefix 与 schema 和表名一起生成 topic 名称。这里 topic 是 shop.public.orders。
  • table.include.list 限制要捕获的表。
  • decimal.handling.mode: string 使 NUMERIC 列以字符串形式发送 "149.90"。Debezium 默认将此值编码为 base64 字节，非常难以阅读。
  • schemas.enable: false 省去每个 JSON 消息附带的 schema 部分，所以 payload 更加精简。

检查连接器状态：

curl -s localhost:8083/connectors/shop-connector/status
{"name":"shop-connector","connector":{"state":"RUNNING","worker_id":"172.18.0.4:8083"},
 "tasks":[{"id":0,"state":"RUNNING","worker_id":"172.18.0.4:8083"}],"type":"source"}

4. 变更事件的结构

添加然后更新一个订单：

docker compose exec postgres psql -U postgres -d shop -c "
  INSERT INTO orders (customer_email, status, total) VALUES ('an@example.com', 'NEW', 149.90);
  UPDATE orders SET status = 'PAID' WHERE id = 1;"

然后用控制台消费者读取 topic：

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic shop.public.orders --from-beginning

UPDATE 事件看起来如下（已简化 source 字段）：

{
  "before": {"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
  "after":  {"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
  "source": {"connector":"postgresql","db":"shop","table":"orders","lsn":24023128, ...},
  "op": "u",
  "ts_ms": 1760170000123
}

op 字段指示变更类型：

op 含义 before after
c INSERT null 新行
u UPDATE 旧行 新行
d DELETE 旧行 null
r 初始快照中的读取 null 现有行

当删除一行时，Debezium 发送一个 tombstone（逻辑删除标记），即具有相同 key 但 value 为 null 的消息。Tombstone 允许 Kafka 日志压缩彻底清理该 key。你的消费者必须能够处理 null 值。

5. 使用 Spring Boot 消费事件

5.1. 依赖项和配置

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

上述依赖项的版本由 spring-boot-starter-parent 3.x 管理。

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-sync
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

这里消费者接收字符串格式的 value，然后自己解析。这种方式帮助你更容易地控制 tombstone 和 JSON 错误，相比使用 JsonDeserializer。

5.2. 模型和解析器

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderRow(
        Long id,
        @JsonProperty("customer_email") String customerEmail,
        String status,
        BigDecimal total) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
        OrderRow before,
        OrderRow after,
        String op,
        @JsonProperty("ts_ms") Long tsMs) {
}

@JsonIgnoreProperties(ignoreUnknown = true) 忽略 source 字段以及 Debezium 以后可能添加的新字段。Jackson 自动将字符串 "149.90" 转换为 BigDecimal，所以你不会像使用 double 那样失去精度。

@Component
public class OrderChangeParser {

    private final ObjectMapper objectMapper;

    public OrderChangeParser(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public Optional<OrderChangeEvent> parse(String value) {
        if (value == null) {
            return Optional.empty(); // tombstone
        }
        try {
            return Optional.of(objectMapper.readValue(value, OrderChangeEvent.class));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Invalid Debezium event", e);
        }
    }
}

解析器为 tombstone 返回 Optional.empty()，所以监听器只需要跳过这个情况。

5.3. 监听器

@Component
public class OrderCdcListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCdcListener.class);
    private final OrderChangeParser parser;

    public OrderCdcListener(OrderChangeParser parser) {
        this.parser = parser;
    }

    @KafkaListener(topics = "shop.public.orders")
    public void onOrderChange(@Payload(required = false) String value) {
        parser.parse(value).ifPresent(event -> {
            switch (event.op()) {
                case "c", "r" -> log.info("Order created: {}", event.after());
                case "u" -> log.info("Order {} status {} -> {}",
                        event.after().id(), event.before().status(), event.after().status());
                case "d" -> log.info("Order deleted: {}", event.before().id());
                default -> log.warn("Unknown op: {}", event.op());
            }
        });
    }
}

@Payload(required = false) 允许 Spring 在遇到 tombstone 时传递 null 到方法。如果缺少这个注解，Spring 会报告转换错误。"r" 分支与 "c" 合并，因为对于消费者，snapshot 中已有的行就像刚创建的行一样。

运行应用程序，使用第 4 节的数据，你将看到：

INFO  OrderCdcListener : Order created: OrderRow[id=1, customerEmail=an@example.com, status=NEW, total=149.90]
INFO  OrderCdcListener : Order 1 status NEW -> PAID

6. 测试解析器

解析逻辑是最容易出错的部分，特别是当 Debezium 或 schema 发生变化时。你可以测试这部分而不需要 Kafka：

class OrderChangeParserTest {

    private final OrderChangeParser parser = new OrderChangeParser(new ObjectMapper());

    @Test
    void parsesUpdateEvent() {
        String json = """
            {"before":{"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
             "after":{"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
             "source":{"connector":"postgresql"},
             "op":"u","ts_ms":1760170000123}
            """;

        OrderChangeEvent event = parser.parse(json).orElseThrow();

        assertThat(event.op()).isEqualTo("u");
        assertThat(event.before().status()).isEqualTo("NEW");
        assertThat(event.after().status()).isEqualTo("PAID");
        assertThat(event.after().total()).isEqualByComparingTo("149.90");
    }

    @Test
    void returnsEmptyForTombstone() {
        assertThat(parser.parse(null)).isEmpty();
    }

    @Test
    void rejectsMalformedJson() {
        assertThatThrownBy(() -> parser.parse("{not json"))
                .isInstanceOf(IllegalArgumentException.class);
    }
}

第一个测试检查 customer_email 列名映射和 BigDecimal 类型。第二个测试确保 tombstone 不会导致监听器出错。第三个测试确认格式不正确的 JSON 会报告明确的错误，而不是被无声地吞掉。

7. 常见错误和何时不应该使用

注意： 复制槽保持 WAL 直到 Debezium 确认已读取。如果连接器停止了很多天，WAL 将会膨胀，可能导致数据库磁盘满。你需要监控 pg_replication_slots，移除连接器时删除槽：SELECT pg_drop_replication_slot('shop_slot');。

你可能会遇到的其他一些错误：

  • 消费者不是幂等的。 Debezium 和 Kafka Connect 保证至少一次。重启后，同一个事件可能到达两次。使用 id 和 source.lsn 来跳过重复，或者设计 upsert 类型的操作。
  • NUMERIC 值变成难以理解的 base64 字符串。 你忘记配置 decimal.handling.mode。同样地，TIMESTAMP 列默认以微秒数形式发送（从 epoch 开始），所以需要自己转换。
  • before 始终为 null。 表还没有设置 REPLICA IDENTITY FULL。
  • 初始快照耗时过长。 对于大表，snapshot.mode=initial 会在流开始前读取整个表。考虑 no_data 如果你只需要从现在开始的更改，或使用增量快照。
  • 数据泄露。 事件反映表的确切结构。如果直接暴露给其他服务，你正在公开数据库 schema。对于业务事件，将 Debezium 与 Outbox pattern 相结合：应用程序在同一事务中写入 outbox 表，Debezium 只捕获那个表。
  • 在生产环境中使用 postgres 账户。 创建一个单独的用户，拥有 REPLICATION 和 SELECT 权限在需要捕获的表上。

你不应该在以下情况使用 CDC：

  • 只有一个应用程序且它可以自己在同一进程中发出事件。此时事务性 outbox 配上一个简单的轮询器就足够了。
  • 你需要具有业务意义的事件，如“OrderPaid”而不是“行 X 更改状态列”。原始 CDC 只知道数据发生了变化，不知道原因。
  • 运维团队还没有准备好管理 Kafka Connect、复制槽和 schema 演进。

8. 总结

在这篇文章中，你已经构建了 PostgreSQL（启用逻辑复制）、Kafka 和 Debezium，然后注册了捕获 orders 表的连接器。你还学习了变更事件的结构，有 before、after 和 op 字段，然后编写了 Spring Boot 消费者来处理 tombstone，并对解析部分进行了测试。

CDC 消除了双写的风险，因为数据库成为唯一的事实来源。作为代价，你必须监控复制槽、编写幂等消费者，并考虑当需要明确的业务事件时使用 Outbox pattern。

kafkadebeziumcdcpostgresqlspring-boot
作者的应用AlterTurn photos into postage stamps, collected in a stamp bookiOS · Android

同主题文章