《Java 消息中间件·可靠消息与消费治理》7.2 CDC：增量快照与变更捕获

CDC 这个词覆盖两种很不一样的东西

变更数据捕获（Change Data Capture，CDC）常被当成一个东西，实际上有两种实现路线，能力边界差得很远：

本篇的实验用手写查询轮询，因为它只用 JDBC 就能完整跑通，能把「水位」「快照衔接」「op 推断」三件事摊开看。日志挖掘路线用 Debezium 做架构对照，本机没有执行，正文会明确标注。

先给结论：查询轮询适合「表结构可控、删除用软删、能接受改代码」的同步场景；日志挖掘适合「不能改业务、要捕获硬删除、要 before-image」的场景。 选错路线，后面补的每个洞都会变成事故。

查询轮询版 CDC 的最小实现

实验表在业务字段之外多了四个 CDC 专用列：

CREATE TABLE cdc_loan (
  loan_id         BIGINT PRIMARY KEY,
  status          VARCHAR(16)  NOT NULL,
  amount_cents    BIGINT       NOT NULL,
  version         BIGINT       NOT NULL,   -- 每次变更递增，全局水位
  created_version BIGINT       NOT NULL,   -- 创建时的 version，用于推断 op=c
  updated_at      TIMESTAMP(3) NOT NULL,
  deleted         BOOLEAN      NOT NULL DEFAULT FALSE
);
CREATE INDEX idx_cdc_version ON cdc_loan(version);

实验跑在嵌入式 H2 2.3.232 上（文件模式，便于跨进程保留数据）。TIMESTAMP(3)、BOOLEAN 这些类型在 MySQL、PostgreSQL 里要换成本地写法，生产中把 JDBC 连接换成你自己的库，CDC 逻辑不变。

轮询的 SQL 只有一句：

SELECT loan_id, status, amount_cents, version, created_version, updated_at, deleted
FROM cdc_loan
WHERE version > ?
ORDER BY version, loan_id;

version 由应用在每次写入时递增，并随行一起提交。用 version 而不是 updated_at 做水位，原因在下一节展开。

op 的推断规则是：

static String op(Row r) {
    if (r.deleted()) return "d";
    if (r.version() == r.createdVersion()) return "c";
    return "u";
}

先 seed 5 行（version 1 到 5），再做三种变更：把 loan 2 的金额改成 2200（version 6）、插入 loan 6（version 7）、软删 loan 3（version 8）。从水位 5 增量拉取，输出正好三条：

[seed] rows=5 version=1..5
[change] update loan=2 amount=2200 version=6 op=u
[change] insert loan=6 amount=6600 version=7 op=c
[change] soft-delete loan=3 version=8 op=d
$ ./run.sh -Dcdc.db=/tmp/mqrel7-cdc/cdc CdcLab incremental 5
[incremental] fromVersion=5 rows=3
[incremental] version=6 loan=2 op=u amount=2200 deleted=false
[incremental] version=7 loan=6 op=c amount=6600 deleted=false
[incremental] version=8 loan=3 op=d amount=3000 deleted=true

三条记录，三种 op。看起来 CDC 的核心就这么点东西。但这份输出藏了三个前提，任何一个不成立，结果就不对。

前提一：op 是推断出来的，不是读出来的

日志挖掘拿到的是一条日志记录，类型字段直接写着 INSERT / UPDATE / DELETE，还带 before 和 after 两个镜像。查询轮询只看到「当前这一行长什么样」，c 和 u 的区别只能靠 created_version 这种额外列推断。

这意味着两件事：

• 业务表要加列。不是零侵入，建表规范和写入代码都要改。
• 推断会失真。如果某次写入忘了更新 version，或者 created_version 被历史数据迁移改过，op 就会判错。日志挖掘没有这个问题。

前提二：硬删除完全不可见

本机把 loan 4 物理删掉，再从水位 8 拉取：

[hard-delete] physical delete loan=4, no row remains
[incremental] fromVersion=8 rows=0
[hard-delete] rows-now=5; snapshot diff would notice, version poll cannot

rows=0。loan 4 从表里消失了，但没有任何一行告诉 CDC「它被删了」。下游会永远保留这条借阅，直到某次全量对账发现多出来一条。

要修只有两条路：

1. 软删除：加 deleted 列，删除变成 UPDATE ... SET deleted=TRUE，如前面的 op=d。
2. 定期全量 diff：快照一遍主键集合，与下游比对，把缺失的键当作删除。这是补偿手段，窗口内的不一致仍然存在。

硬删除是查询轮询路线的硬边界，不是实现细节。

前提三：水位必须单调，且不能有并列

version 由应用维护，就要保证：同一事务内多次修改只产生一个版本号，且版本号在提交顺序上单调。如果用数据库序列在事务开始时就取号，两个并发事务可能先取到大号、后提交，CDC 会先看到大号再看到小号，水位一推进，小号那条就永远丢了。

安全的做法是让版本号在提交时生成（例如同表 MAX(version)+1 加行锁，或单写者），或者干脆用数据库的提交序列（如 PostgreSQL 的 LSN）。这也是为什么日志挖掘更省心：日志顺序就是提交顺序。

水位列：为什么 version 比 updated_at 可靠

很多团队用 updated_at > last_update_time 做增量。这个写法有两个坑，本机都复现了。

坑一：同一毫秒的多行会漏

插入两行，updated_at 完全相同（2023-11-15 06:13:29.0）。处理完第一行后把水位推进到这个时间，再查 updated_at > 水位：

[same-ts] ts=2023-11-15 06:13:29.0 single-column-rows-after-watermark=0 composite-rows=1

单列水位查到 0 行，第二行被跳过。改成复合水位 (updated_at, loan_id)，查询条件写成：

WHERE updated_at > ?
   OR (updated_at = ? AND loan_id > ?)

同样的数据查到 1 行。复合水位的关键是排序键与比较键必须一致：按 (updated_at, loan_id) 排序，就用这个二元组做游标；只按时间比较，时间相同的行必然丢一半。

坑二：时间戳不等于提交顺序

updated_at 通常是应用在写入时取的 System.currentTimeMillis()。两个事务并发：

轮询按时间排序，会先读到 A 的时间 100，把水位推到 100；B 的时间是 200，还没被读到。等 B 提交后下一轮扫描，updated_at > 100 能查到 B，没问题。真正危险的是反过来：B 先提交但时间更大，A 后提交但时间更小。轮询先看到 B，水位推到 200，A 的 100 永远落在水位之下，直接丢失。

这不是理论问题：应用服务器时钟回拨、NTP 校正、批量导入使用业务时间，都会制造这种倒序。version 由数据库在写入路径上单调分配，没有这个问题；日志挖掘只依赖数据库自己的日志序列，不依赖应用时钟。

三种水位的取舍

快照与增量的衔接

CDC 上线时，表里已经有历史数据。只读增量不够，要先做一次快照，再从某个水位接上增量。衔接点选错，就会丢数据。

正确顺序是：先取水位，再做快照，最后从水位拉增量。原因是快照读到的行可能已经包含了水位之后的更新，重复可以用版本号去重；但如果先做快照、后取水位，快照读完之后、取水位之前发生的更新，既不在快照里（快照读的是旧值），也不在增量里（水位已经越过它），直接丢失。

本机把两种顺序都跑了一遍。场景相同：快照读到 loan 1 的旧值 1000，随后 writer 提交 amount=9999, version=6。

先取水位：一致

[snapshot] watermark-before-scan=5
[snapshot] read loan=1 amount=1000
[snapshot] read loan=2 amount=2000
[snapshot] read loan=3 amount=3000
[snapshot] read loan=4 amount=4000
[snapshot] read loan=5 amount=5000
[race] writer committed loan=1 amount=9999 version=6
[incremental] from=5
[apply] loan=1 amount=9999 version=6
[result] applied=1 state.loan1=9999 expected=9999 consistent=true

快照读到了旧值 1000，但水位是 5，增量从 5 开始，把 version 6 的更新补上，最终状态 9999。快照里的旧值被增量覆盖，结果正确。

后取水位：丢失

[snapshot] read loan=1 amount=1000
[snapshot] read loan=2 amount=2000
[snapshot] read loan=3 amount=3000
[snapshot] read loan=4 amount=4000
[snapshot] read loan=5 amount=5000
[race] writer committed loan=1 amount=9999 version=6
[snapshot] watermark-after-scan=6
[incremental] from=6
[result] applied=0 state.loan1=1000 expected=9999 consistent=false

快照读的是旧值 1000，水位却是 6，增量从 6 开始查不到任何东西，最终状态停在 1000。这一行在快照和增量之间「消失」了，而且没有任何报错。

衔接的三条规则

1. 水位在快照开始前取，并且与快照读在同一个一致性视图里（同一事务或同一快照隔离级别）。取水位和读数据分成两个事务，中间仍可能有更新。
2. 增量按主键去重：快照里可能已经包含水位之后的版本，应用增量时比较「事件版本 > 本地已存版本」再写入，重复就跳过。上面的 consistent=true 依赖的正是这条规则。
3. 快照期间不推进水位。水位是快照开始时的边界，不是「当前最大版本」；增量任务从边界开始追，追上后再切换到实时位点。

增量快照的分块形态

全表快照有两个现实问题：大表一次读完会占满 IO 或长事务，中途失败要从头再来。生产级 CDC 通常把快照切成块，每块单独提交，失败只重做当前块。

分块快照的骨架（这是架构描述，不是本机执行的代码）：

1. 记录 low watermark = 当前最大版本
2. 按主键范围把表切成 N 块：[min, k1], (k1, k2], ...
3. 逐块 SELECT ... WHERE pk > ? AND pk <= ? ORDER BY pk
4. 每块读完提交，记录进度；块之间继续消费增量并缓存
5. 全部块完成后，记录 high watermark，把缓存与增量按主键去重应用
6. 从 high watermark 切入实时流

关键点还是去重：块快照读到的行可能比缓存里的增量更新，应用时按主键比较版本，取版本大的那个。顺序不重要，版本号重要。

Debezium 的 incremental snapshot 就是这个形态：通过信号表触发，记录 low/high watermark，分块读，块之间用日志流缓冲变更，最后按主键去重。具体实现没有在本机执行，这里只作架构对照。

op 语义：c/u/d 从哪里来

op 语义直接影响下游。比如缓存失效：u 和 d 都要删缓存，但 c 可以只写不删；如果查询轮询把一个新插入误判成 u，下游可能多做一次无谓的删除，通常无害。反过来，把 u 误判成 c，下游如果对 c 做了「不存在才插入」的处理，就会跳过这次更新，缓存永远停在旧值。

所以查询轮询版 CDC 的事件体里，除了 op，还应该带上 version 和 updated_at。下游可以用版本号做单调性判断，不必完全相信 op。

与 outbox 的分工

Outbox 和 CDC 经常被放在一起比较，其实它们解决的是不同层的问题：

一个常见的组合是：业务写 outbox 表，CDC 只订阅 outbox 表，把「轮询发布」换成「日志捕获发布」。这样事件语义仍由业务定义，relay 的轮询压力也没了，代价是引入一个 CDC 组件和它自己的位点、schema 历史、故障恢复。

如果表结构简单、团队不想引入连接器，纯轮询 outbox 也完全够用，7.1 的实现就是证明。选型要看的是「事件语义」和「运维预算」，不是哪个听起来更先进。

Debezium 架构对照（未在本机执行）

Debezium 是日志挖掘路线的代表实现。它的组件与数据流如下：

它的投递语义默认是至少一次：连接器崩溃重启后从上次位点重放，可能产生重复，下游按主键去重。开启 Kafka 事务后可以做到「读—处理—写」的恰好一次，但那是另一套成本，且仍然不覆盖外部副作用。

部署形态通常是一个 Kafka Connect 集群加一个 connector 配置。下面是命令与配置的预期形态，本机未执行，不作为可复现实验：

docker run -d --name connect -p 8083:8083 \
  -e BOOTSTRAP_SERVERS=kafka:9092 \
  -e GROUP_ID=cdc-lab \
  -e CONFIG_STORAGE_TOPIC=connect-configs \
  -e OFFSET_STORAGE_TOPIC=connect-offsets \
  -e STATUS_STORAGE_TOPIC=connect-status \
  debezium/connect:<按你的实测版本>

curl -X POST http://127.0.0.1:8083/connectors -H 'Content-Type: application/json' -d '{
  "name": "libraryd-cdc",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "cdc",
    "database.password": "***",
    "topic.prefix": "libraryd",
    "table.include.list": "libraryd\\.loan",
    "snapshot.mode": "initial"
  }
}'

预期结果是 POST 返回 201，GET /connectors/libraryd-cdc/status 的 connector.state 为 RUNNING，libraryd.libraryd.loan 主题出现带 op、before、after 字段的 change event。这些输出没有在本机取得，因为本机没有部署 MySQL 与 Debezium 连接器；正文只用它说明组件关系与配置项，不声称跑通。

需要强调：Debezium 的配置项随版本变化，database.history 一类旧参数在新版本已被 schema history topic 取代。真要上线，以对应版本文档和实测为准，不要照抄博客。

本节判据

1. 日志挖掘和查询轮询最本质的区别是什么？ 答：日志挖掘拿到的是带 before/after 的变更记录，查询轮询只看到当前行，op 与删除都要靠额外列推断。
2. 为什么用 updated_at 做水位会漏数据？ 答：同一毫秒的多行会被 > 跳过；应用时钟回拨或并发提交还会让时间顺序与提交顺序相反。
3. 快照与增量应该按什么顺序衔接？ 答：先取水位，再做快照，最后从水位拉增量；反过来会丢掉快照读完之后、取水位之前发生的更新。
4. 查询轮询版 CDC 为什么看不见硬删除？ 答：行被物理删除后没有任何可轮询的痕迹，只能靠软删列或定期全量 diff 补偿。
5. Outbox 与 CDC 该怎么选？ 答：要业务语义清晰、跨服务契约就用 outbox；不能改业务、要捕获硬删除和 before-image 就用日志挖掘 CDC，两者也可以组合成「outbox 表 + CDC 订阅」。

阅读导航：上一节 7.1 Outbox 模式 ｜ 下一节 7.3 事件溯源入门