怎么用这篇附录
这四份模板不是四个「最佳实践」，而是四种不同前提下的解。选错模板比不选更糟：用状态机去处理「通知发送」这种没有实体状态的事件，会写出一个永远推进不了的假状态机；用唯一约束去处理「同一实体一天多条」的事件，会把合法消息当重复丢掉。
每份模板都标了三件事：
- 解决哪一类重复——这个模板对哪种重复有效。
- 窗口在哪——它没有关掉的那一段，什么时候会漏。
- 接哪一节——正文里对应哪一章的展开。
代码用 H2 的语法，DataSource 与 KafkaProducer 的具体实现按你的技术栈替换。所有模板都只依赖 JDBC 与 kafka-clients，不引入连接池或 ORM。
模板一：去重表
适用：任意事件，尤其是没有实体状态、无法用状态机表达副作用的（发短信、调外部接口、写日志流水）。
解决的重复：所有来源的重复投递——生产端重试、再平衡、位移回退、上游重放。
窗口：去重记录与业务写入之间。两者必须同一个事务，否则会出现「记了去重但业务没做」或反过来。
CREATE TABLE consumed_event (
event_id VARCHAR(64) PRIMARY KEY,
consumer VARCHAR(64) NOT NULL,
consumed_at TIMESTAMP NOT NULL,
payload_hash VARCHAR(64) NOT NULL
);
consumer 列不能省。同一份事件可能被多个消费组各处理一次，去重表的键必须是「事件 + 消费者」，否则第二个组的处理会被第一个组误判为重复。
boolean handleOnce(Connection conn, String eventId, String consumer,
Runnable businessAction) throws SQLException {
try (PreparedStatement ps = conn.prepareStatement(
"INSERT INTO consumed_event(event_id, consumer, consumed_at, payload_hash) " +
"VALUES (?, ?, CURRENT_TIMESTAMP, ?)")) {
ps.setString(1, eventId);
ps.setString(2, consumer);
ps.setString(3, hashOf(payload));
ps.executeUpdate();
} catch (SQLException e) {
if (e.getSQLState() != null && e.getSQLState().startsWith("23")) {
return false; // 唯一键冲突 = 已经处理过
}
throw e;
}
businessAction.run(); // 与上面的 INSERT 在同一个事务里
return true;
}
三个要点：
- 靠 SQLState 判断重复，不靠异常类名。H2 抛
JdbcSQLIntegrityConstraintViolationException，PostgreSQL 抛PSQLException，捕获具体类名会让代码绑死在一个数据库上。SQLState 以23开头是 SQL 标准的「完整性约束违反」。 businessAction必须用同一个conn。换一个连接就是另一个事务，原子性消失。- 外面的调用方管事务。这个方法只做「插入 + 业务」，
commit或rollback在上层，否则「插入成功、业务失败」时无法回滚。
TTL 与容量：去重表会随事件数线性增长。清理策略要么按 consumed_at 定期删，要么按分区归档。删除的时间窗必须大于「消息可能被重放的最大延迟」，否则重放时会重新消费。这个窗口的下界通常取消息保留期——如果 Agent 的重放可能来自 7 天前的数据，去重记录就得留 7 天以上。
模板二：状态机
适用：有明确生命周期的实体（订单、借阅、工单、账户）。
解决的重复：同一事件重复投递、不同事件的乱序到达。
窗口：状态读取与更新之间。必须用条件更新或行锁，否则两个并发消费者会各自读到旧状态、各自推进。
enum LoanState { CREATED, APPROVED, RETURNED, CANCELLED }
private static final Map<LoanState, Set<LoanState>> ALLOWED = Map.of(
LoanState.CREATED, EnumSet.of(LoanState.APPROVED, LoanState.CANCELLED),
LoanState.APPROVED, EnumSet.of(LoanState.RETURNED)
);
迁移判断的关键在于区分「重复」与「非法」：
| 当前状态 | 事件目标状态 | 处理 |
|---|---|---|
| CREATED | APPROVED | 执行迁移 |
| APPROVED | APPROVED | 幂等成功，不迁移、不报错 |
| APPROVED | CREATED | 抛异常，说明乱序或上游有 bug |
| RETURNED | 任何 | 抛异常，终态不可迁移 |
第二行是最容易被写成 bug 的地方：把「目标状态等于当前状态」当作正常返回，而不是当作重复。它在重复投递时表现正确，但在「同样的两条合法事件」时也会静默吞掉第二条——如果业务允许同一实体多次进入同一状态（比如多次逾期），那就不能用状态机，回到模板一。
条件更新是防并发的手段：
UPDATE loan
SET status = ?, updated_at = CURRENT_TIMESTAMP
WHERE loan_id = ?
AND status = ? -- 期望的旧状态
检查返回的影响行数：0 行说明状态已被别人改过，重读后重新判断。这个「读—判断—条件更新」的循环最多重试两次，第三次还失败说明该实体的消息在乱序或重复投递，应当告警而不是无限重试。
模板三：outbox
适用：本地数据库要改，同时必须保证对应的领域事件一定发出去。
解决的重复：不是重复，是丢失。它保证「业务行提交了，事件就一定最终会发」。
窗口：relay 发送成功与标记 published 之间。这段窗口会产生重复投递，需要消费端幂等（模板一或模板二）兜住。outbox 从来不承诺「恰好一次」，只承诺「至少一次」。
表结构与 7.1 节一致，这里重复列出以便直接复制：
CREATE TABLE business_loan (
loan_id VARCHAR(32) PRIMARY KEY,
book_id VARCHAR(32) NOT NULL,
member_id VARCHAR(32) NOT NULL,
status VARCHAR(16) NOT NULL,
created_at TIMESTAMP NOT NULL
);
CREATE TABLE outbox (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
event_id VARCHAR(64) NOT NULL UNIQUE,
aggregate_id VARCHAR(32) NOT NULL,
event_type VARCHAR(64) NOT NULL,
payload CLOB NOT NULL,
created_at TIMESTAMP NOT NULL,
published BOOLEAN NOT NULL DEFAULT FALSE
);
业务写入与 outbox 写入在同一个事务里：
try (Connection conn = ds.getConnection()) {
conn.setAutoCommit(false);
try {
insertLoan(conn, loan);
insertOutbox(conn, eventId, loan.loanId(), "loan.created", payload);
conn.commit();
} catch (SQLException e) {
conn.rollback();
throw e;
}
}
relay 用轮询，按 id 顺序取未发布的行：
List<OutboxRow> batch = selectUnpublished(conn, 100);
for (OutboxRow row : batch) {
ProducerRecord<String, String> rec =
new ProducerRecord<>(topicFor(row.eventType()), row.aggregateId(), row.payload());
rec.headers().add(new RecordHeader("event-id",
row.eventId().getBytes(StandardCharsets.UTF_8)));
producer.send(rec, (meta, err) -> {
if (err == null) markPublished(row.id());
// 失败不标记，下一轮重试
});
}
producer.flush();
四个必须注意的点：
event_id在建表时就生成并写进 outbox，不要每次 relay 重发时重新生成。重发时必须用同一个eventId，否则消费端去重失效。- 先
send成功再标记published。反过来会丢。顺序错了这一条是整个模式失效的唯一入口。 - 多个 relay 实例并发会重复发送。要么用单实例，要么用
SELECT ... FOR UPDATE SKIP LOCKED抢占。无论哪种，消费端幂等都是必须的。 - 表会持续增长。已发布的旧行要定期归档，
published列上要有索引，否则轮询会扫全表。
模板四：事务生产者
适用：消费—处理—生产这条链路要求原子——要么「读了这批消息、产出了那批消息、位移推进」三件事一起成功，要么一起不发生。
解决的重复：消费端崩溃导致的「处理了但位移没提交」的重复，以及「产出了但位移没提交」导致的下一轮重复生产。
窗口：外部副作用。事务只能覆盖 Kafka 内部的写入（产出消息 + 位移），覆盖不了数据库写入、HTTP 调用。这是它最常被误解的边界。
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "libraryd-loan-tx-1");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "60000");
producer.initTransactions();
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(cProps)) {
consumer.subscribe(List.of("libraryd.loan"));
while (running) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
producer.beginTransaction();
try {
for (ConsumerRecord<String, String> r : records) {
producer.send(new ProducerRecord<>("libraryd.audit", r.key(), r.value()));
}
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
for (TopicPartition tp : records.partitions()) {
long next = records.records(tp).getLast().offset() + 1;
offsets.put(tp, new OffsetAndMetadata(next));
}
producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}
}
}
五个要点：
TRANSACTION_TIMEOUT_CONFIG不是..._MS_CONFIG。ProducerConfig里这个常量就叫TRANSACTION_TIMEOUT_CONFIG，写成TRANSACTION_TIMEOUT_MS_CONFIG编译不过。- 位移要
+1。OffsetAndMetadata存的是「下一条要读的位置」，不是「最后处理的位置」。写错会导致重复消费或无端跳过一条。 transactional.id必须跨实例稳定。同一个逻辑消费者重启后要用同一个 id，否则 fencing 失效、两个实例同时写。它是身份，不是实例名。- 下游消费者必须
read_committed。read_uncommitted会看到被 abort 的消息，事务白做。3.1 节实测过两种隔离级别看到的结果差异。 - 数据库写入不在事务内。要保证「写库 + 发消息」原子，用模板三（outbox），不要试图用 Kafka 事务覆盖数据库——两者是两个独立的事务管理器。
按场景选模板
| 场景 | 模板 | 理由 |
|---|---|---|
| 创建型事件、键天然唯一 | 唯一约束（业务表主键） | 成本最低，无需额外表 |
| 无状态副作用（通知、调用外部接口） | 模板一 去重表 | 状态机表达不了 |
| 有生命周期的实体 | 模板二 状态机 | 顺带解决乱序 |
| 本地库改动 + 必须发事件 | 模板三 outbox | 消除「有业务无事件」 |
| 消费—处理—产出 要原子 | 模板四 事务生产者 | 位移与产出一起提交 |
| 跨库/跨服务的多步业务 | 9.1 的 Saga | 补偿而非回滚 |
组合是常态而不是例外：一个真实的链路往往是「模板四做 Kafka 内的原子 + 模板一兜住 relay 的重复 + 模板三保证业务事件不丢」。三种叠加不矛盾，因为它们覆盖的是三个不同的窗口。
怎么测这四份模板
每个模板都有一个「最小可验证断言」，写完立刻能验，不要等到集成测试：
| 模板 | 断言 | 期望 |
|---|---|---|
| 去重表 | 同一个 eventId 调两次 handleOnce | 第一次 true、第二次 false，业务表只多一行 |
| 状态机 | APPROVED → APPROVED 与 APPROVED → CREATED | 前者返回幂等成功，后者抛异常 |
| outbox | 事务内抛异常后查 outbox | 业务行与 outbox 行都不存在 |
| outbox relay | 发送成功后查 published | 变为 TRUE；发送失败则保持 FALSE 等下轮 |
| 事务生产者 | abort 后用两种隔离级别消费 | read_committed 看不到，read_uncommitted 看得到 |
最后一行同时验证了模板与配置两件事。如果 read_committed 也能看到被 abort 的消息，先查下游消费者的 isolation.level，而不是改事务代码。
四份模板共同的三个坑
幂等键必须在生产端生成。 四份模板全都依赖一个「重试时不变」的键。如果生产端每次重试都换一个 eventId，下游所有去重都失效。这是唯一一个无法在下游补救的问题。
TTL 不能短于重放窗口。 去重表、outbox 归档、状态机的终态清理，三者的时间窗都要覆盖「消息可能被重放的最大延迟」。这个延迟的上界通常就是消息保留期。
模板不解决「处理成功但副作用失败」。 去重表记了、业务写入回滚了，重试时会被判重复。所以 businessAction 里不能有不可回滚的外部调用——调用外部接口必须走「先记录意图、再异步执行、按结果标记」的路径，也就是 outbox 的思路。这一点在 6.3 节讨论毒消息时也出现过：不可回滚的副作用，是重试机制的天敌。
配置矩阵见附录 A，机制展开见第 2、3、7、9 章。这两篇附录不含新概念，只在抄代码时翻。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。