7 队列、流与事件驱动

第 6 章的最后停在 outbox：业务行和消息行在同一个本地事务里提交，消费端用去重表挡住重复。那还是在「一次请求、一个数据库」的范围内。本章把镜头拉远：一次状态变化被写成事件之后，它会被多少个消费者、以什么顺序、重复多少次地读到？

一个支付成功事件 payment.charged 通常有四个消费者：记账、通知、风控、数据仓库。记账 consumer 停机两小时，恢复后要从两小时前的位置继续；通知 consumer 发版后从旧 offset 重放，可能给用户再发一条短信；风控要求同一张卡的事件严格有序；数仓只关心最终结果，重复和乱序都能靠聚合消化。同一条事件，四种「次数」和「顺序」要求。

队列和日志是两种不同的答案。队列把消息当任务，ack 之后删除；日志把消息当事实，追加之后保留，消费者用 offset 表示自己读到哪。选错模型，后面所有补偿都是在给语义打补丁。本章先讲清这两种模型，再讲投递语义、分区与消费者组、offset 提交、Kafka 的复制与事务，最后落到 outbox、CDC、事件溯源，以及一条 offset 提交失败导致的二次扣款。

路线是：7.1 队列与日志；7.2 三种投递语义；7.3 分区、顺序与消费者组；7.4 offset 提交与再平衡；7.5 ISR、acks、幂等生产者与事务；7.6 outbox、CDC 与事件溯源。后半程是四个必须单独成节的事：重复投递与乱序（7.7）、consumer lag 与 rebalance 的观测（7.8）、LinkedIn 的 Kafka 规模与运维（7.9），以及 offset 提交失败后的二次扣款（7.10）。最后是事件设计清单、陷阱和出口检查。

一句话先放在这里：消息系统的语义不是配置出来的，是生产者、broker、消费者三方行为组合出来的。

7.1 队列与日志：删除 vs 保留

先看两个方向相反的工单。

工单 A：一个发邮件的任务队列，消息的可见性超时是 30 秒，处理函数偶尔要跑 45 秒。超时后 broker 把同一条消息投给另一个 worker，两个 worker 都发了邮件。用户收到两封。

工单 B：一个 Kafka 消费者组停机了五天，topic 的保留期是三天。恢复时 consumer 的 offset 已经落在日志起点之前，客户端报 OffsetOutOfRange，按策略重置到 latest，五天的事件被跳过。下游报表少了一周的数据。

一个多了，一个少了，但问的是同一个问题：消息被消费之后，还应该存在多久？ 队列的回答是「ack 就删」，日志的回答是「到期才删」。这个差异决定了消费进度归谁管、能不能重放、慢消费者会看到什么。

这张表里最重要的是第三行：队列的进度在 broker，日志的进度在消费者。队列可以在超时后把消息再给你一次，因为它知道你没 ack；日志不会主动找你，它只在你回来时告诉你「你上次说到 4，但日志现在从 100 开始了」。

日志里的 offset 是一个 64 位整数，表示记录在分区内的位置。记录一旦写入，offset 就固定；新记录追加在末尾。消费者提交的 committed offset 是「下一条要读的位置」，不是「最后一条读过的位置」。如果处理完 offset 5 之后提交，提交的值是 6。

图中消费者处理到 msg 5，但只提交到 4。按图里的「最后处理位点」口径，4 及之前已确认，重启从 5 开始读，msg 5 被再处理一次。Kafka API 的 committed offset 是「下一条要读」，同一个状态提交的值是 5；两种口径差一位，读监控、日志和文档时必须先确认用的是哪一种，否则会多算或少算一条。「处理成功」和「提交成功」是两个动作，两者之间有一个无法消除的窗口；这个窗口就是本章后面所有去重逻辑的起点。

还有三个位置需要区分，它们经常被混为一谈：

LEO - committed offset 就是常说的 lag。它不是「有多少条没处理」，而是「有多少条还没提交」；处理了但没提交的记录也算在 lag 里。这个区别在 7.8 会用到。

保留策略不是一个运维参数，而是日志对外契约的一部分。Kafka 的 retention.ms/retention.bytes 按段（segment）删除，段文件名就是该段的 base offset，例如 00000000000000000123.log 表示这个段从 offset 123 开始。删除以段为单位，所以实际保留时间可能比配置多出一个段。消费者停机时间一旦超过保留期，恢复时就会遇到 OffsetOutOfRange；此时 auto.offset.reset 决定它是跳到最早还是最新，两种都会留下数据缺口。

日志还有另一种保留方式：压实（compaction）。它不按时间删除，而是对每个 key 只保留最后一条记录，更早的版本在后台被清理；删除某个 key 则写一条 value 为空的墓碑（tombstone），过一段时间连墓碑也被清理。压实适合「用日志保存一张表的当前状态」，例如 Kafka 里保存消费者偏移的 __consumer_offsets，或者事件溯源里的快照流。压实之后 offset 会出现空洞，所以 offset 不能当消息 ID 用。

队列也有「保留」，只是表达方式不同：可见性超时、消息 TTL、死信队列。它们的共同点是由 broker 决定消息还能不能被看到；日志则把「还能看多久」写进了保留策略，消费者必须自己保证在窗口内读完。

7.1.1 什么时候该选哪一个

两种模型没有优劣，只有匹配。把业务问题压缩成五个问题，答案会指向其中一边：

典型的错配是把「任务」放进日志：任务要求「一个任务只执行一次」，日志却允许多个消费者组各读一遍；要做对，必须给任务加唯一 ID、去重表和完成状态，等于在日志上重建队列。反过来，把「事件流」放进队列：事件需要回放给新消费者、需要保留一周做审计，队列 ack 就删，做不到；只能靠 DLQ 和额外的归档表拼出来。

一个折中的判断方法是问「这条消息的价值在消费之后是否还存在」。任务完成后价值就结束了，队列合适；事件是事实，消费之后仍然对审计、重放、新消费者有价值，日志合适。如果一条消息既需要「只做一次」又需要「永久可回放」，通常要用日志 + 幂等消费，而不是队列 + 归档。

7.2 投递语义：最多一次、至少一次、有效一次

讨论「消息会不会丢」之前，先接受一个事实：发送方无法区分「消息没到」和「消息到了但确认丢了」。这是第 1 章 unknown 状态在消息系统里的翻版。面对这个不确定性，只有两种选择：

• 不重试：最多一次。确认丢了就丢了，至少不会重复。
• 重试：至少一次。确认丢了就再发，重复由接收方处理。

「恰好一次投递」在不可靠网络上做不到；工程上说的 exactly-once 指的是恰好一次效果（effectively once）：消息可能被投递多次，但副作用只发生一次。这个区别不是文字游戏，它决定了你在哪一层写去重代码。

表中「先提交再处理」和「先处理再提交」只差一行代码，语义却相反。很多团队说「我们用的是至少一次」，代码却是自动提交 + 异步处理，实际是最多一次：offset 已经提交，异步任务还在队列里，进程一挂就丢了。

重复不只来自消费者。把一条消息从产生到消费拆开，至少有七个位置会产生重复：

丢失的位置同样多：acks=0/1 时 leader 崩溃、自动提交早于处理、捕获异常后继续提交、保留期短于停机时间、生产者缓冲区满后丢弃、进程退出前没有 flush。投递语义是三方行为的组合，不是 broker 的一个开关。

一个常见的误解是「Kafka 是至少一次，所以我的应用也是至少一次」。Kafka 只保证「在保留期内，已提交 offset 之后的记录还能读到」；consumer 完全可以用自动提交把它变成最多一次，也可以用「处理成功但提交失败」把它变成重复。broker 提供的是原材料，语义是应用组装出来的。

# 最多一次：先提交，再处理
for msg in consumer:
    consumer.commit(msg.offset + 1)
    handle(msg)              # 这里崩溃 -> 消息永远不会再来

# 至少一次：先处理，再提交
for msg in consumer:
    handle(msg)              # 这里崩溃 -> 消息会从旧 offset 重来
    consumer.commit(msg.offset + 1)

两段代码的差别只有一行，但第二段要求 handle 是幂等的，否则「至少一次」就变成「至少两次副作用」。第 1 章的幂等键、第 6 章的去重表，在这里重新出现；消息系统只是把这个问题从「一次 RPC」放大成「一条流」。

7.2.1 重复不是小概率

「重复率很低」是很多团队不做幂等的理由。把绝对值算出来，这个理由通常站不住。假设一个 topic 每天 1 亿条事件，处理成功率 99.9%，每天有 10 万条失败；失败重试和 rebalance 重放加起来，只有 1% 的处理会产生重复效果，也是每天 1000 次。对支付来说，1000 次重复扣款不是「小概率」，是事故。

rebalance 会把重复集中放大。一个每秒 1 万条的分区，滚动重启触发 5 分钟的重放窗口，涉及约 300 万条记录；如果其中 0.1% 的处理器不幂等，就是 3000 次重复。重复的规模取决于「一次重放涉及多少条」和「处理器有多不幂等」，而不是取决于「重复率」。

这张表解释了两件事。第一，重放的批次越大，幂等越重要；单条重试时可以靠人工发现，百万级重放只能靠代码。第二，去重表的容量要按「最大重放窗口」估算，而不是按「日常重复率」估算；如果去重记录只保留 7 天，而一次补数重放跨越 30 天，前面的重复就挡不住。

7.3 分区、顺序与消费者组

「Kafka 只保证分区内有序」这句话几乎每篇文档都有，但真正要回答的是两个工程问题：你的分区键是什么？这个键会不会变？

分区是 Kafka 里所有事情的单位：顺序、并行度、复制、保留、压实、offset 都以分区为边界。一个 topic 有 N 个分区，生产者按 key 把记录映射到其中一个：

partition = hash(key) % num_partitions

同一个 key 在分区数不变时总是落到同一个分区，因此同一 key 的记录按写入顺序排列。跨 key、跨分区没有全局顺序；消费者组里不同消费者各自处理不同分区，也无法知道另一个分区读到哪。

分区键的选择直接决定「顺序保证的范围」。如果支付事件用 user_id 做 key，同一用户的事件有序，不同用户之间无序；如果风控需要「同一张卡」有序，就必须用 card_id，而不能指望 user_id。如果两个事件需要相互比较先后（比如 order.created 和 order.cancelled），它们必须用同一个 key，通常是 order_id。

一个经常被忽略的事实是：增加分区会改变 key 到分区的映射。hash(key) % 4 和 hash(key) % 5 对同一个 key 通常给出不同结果，原来在同一分区里的两个事件会被拆到两个分区，顺序保证当场断裂。Kafka 不能减少分区，增加分区也不迁移已有数据；已经写入的记录留在旧分区，新记录按新映射走。于是同一 key 的历史和未来可能落在不同分区，消费者看到的是两段各自有序、彼此无序的流。

这件事的工程含义是：分区数要在建 topic 时按未来一到两年的峰值定，不能等不够了再加。如果确实要扩，正确做法是接受「迁移窗口内该 key 无序」，在消费端用版本号/序号拒绝旧事件，或者双写新旧 topic 做一次受控迁移。直接把分区数从 4 改成 8，然后假设顺序还在，是很多「偶发乱序」工单的根因。

topic payments.charged（4 个分区）

P0: [offset 100][101][102] ...   ──▶  consumer C1（ledger-1）
P1: [offset 200][201] ...        ──▶  consumer C1
P2: [offset 300][301][302] ...   ──▶  consumer C2（ledger-2）
P3: [offset 400] ...             ──▶  consumer C2

group ledger：C1 拥有 P0/P1，C2 拥有 P2/P3
group notify：C3 拥有 P0–P3（不同组，独立 offset）

消费者组（consumer group）是分区分配的容器。同一个组内，一个分区同时只能被一个消费者拥有；不同组之间互不影响，各自读到全量消息、各自维护 offset。这就是为什么「加一个消费者」有时能扩容、有时只是多一个空闲进程：并行度的上限是分区数。

热点分区值得单独说。假设一个 topic 有 12 个分区，key 是 tenant_id，其中一个租户占了 60% 的流量，那么这个分区会一直 lag，其他 11 个分区闲着。加消费者没用，因为瓶颈是一个分区；加分区也没用，因为同一个租户仍然落在一个分区。可选的做法是：给热 key 加后缀拆成多个子分区（但会破坏该租户的全局顺序）、给大租户单独建 topic、或者在业务层做预聚合。没有一种做法能同时保住「单 key 全局有序」和「无限并行」，必须选一个。

rebalance 的触发条件比想象中多：消费者加入/离开、消费者崩溃、max.poll.interval.ms 超时（处理一批太久）、订阅的 topic 分区数变化、订阅关系变化、coordinator 切换。每次 rebalance 都意味着分区所有权转移，未提交的进度可能被重放。7.4 会把这条时间线拆开。

7.3.1 分区数怎么估

分区数不是拍脑袋定的，它同时受三个下界约束：

partitions >= ceil(峰值写入 MB/s / 单分区安全写入 MB/s)
partitions >= 期望的消费者并行度
partitions >= 峰值 key 基数 / 单分区可接受的 key 数   # 防止热分区

第三个下界经常被忽略。如果 key 的基数只有 100，分区开到 64，看起来并行度很高，实际上大部分 key 还是会挤在少数分区；反过来，key 基数很大但流量集中在前 1% 的租户，也要按「最热 key 的流量」而不是「平均流量」来估。

一个算例：峰值写入 60 MB/s，单分区在 acks=all、3 副本、压缩后能稳定吃 8 MB/s，消费者希望 12 个并行，那么：

max(ceil(60 / 8), 12) = max(8, 12) = 12
加 30% 余量：ceil(12 * 1.3) = 16

最终定 16 个分区。这个数字要写进容量文档，和「为什么不是 8、不是 64」一起评审。单分区 8 MB/s 不是常数，它取决于消息大小、压缩率、副本同步、磁盘和网络；用自己的流量压测一遍，不要抄别人的数字。

分区数的调整只有「增加」一个方向，而且增加的那一刻改变 key 映射。所以它更像「建表时选主键」，而不是「运行时调线程池」；宁可一次多留 30% 余量，也不要在业务高峰做重映射。

7.4 Offset 提交与再平衡

offset 提交是消费者唯一能在崩溃后留下的东西。处理结果在数据库里，进度在 coordinator 里；两者不在一个事务里，就一定有窗口。

先看自动提交。enable.auto.commit=true 时，客户端每隔 auto.commit.interval.ms（默认 5 秒）提交一次，提交的是最近一次 poll 返回的记录的末尾，而不是「已经处理完的记录」。如果处理是异步的、或者一批记录处理到一半失败，自动提交可能把没处理完的记录也提交掉：

poll -> [100, 101, 102, 103]
处理 100 成功，101 失败
自动提交 offset=104        # 102、103 永远不会再被这个组读到

这就是「自动提交 + 异步处理」等于最多一次的原因。自动提交不是错，错在把它和处理进度当成同一件事。

手动提交把控制权拿回来，但有三种写法，语义不同：

异步提交有一个隐蔽的坑：commitAsync(offset=104) 失败后，下一次 commitAsync(offset=110) 成功，104–109 就被跳过了。异步提交不会自动重试失败的那一次；后一次成功的大 offset 会覆盖前一次，中间段就永远不会被重放。所以失败的那一批只能靠业务幂等或重新处理来兜底；在 rebalance 的回调里必须用同步提交，否则分区已经转走，异步提交会直接失败。

再平衡把这个问题放大。考虑下面这条时间线：

t0  C1 拥有 P0，已提交 offset=100，poll 到 100..109
t1  C1 处理 100（扣款成功），101..109 还在本地队列
t2  发版/GC/下游慢，C1 超过 max.poll.interval.ms 没有 poll
t3  coordinator 认为 C1 失联，generation 42 -> 43，P0 分配给 C2
t4  C1 处理完 109，调用 commitSync(110)
t5  coordinator 拒绝：generation 已变（CommitFailedException）
t6  C2 从已提交 offset=100 开始，100..109 全部重放
t7  如果 handle 不幂等，100 被扣第二次

注意 t4 的提交被拒绝，不是因为网络，而是因为 C1 已经是一个「僵尸消费者」：它失去了分区所有权，却还在处理旧数据。Kafka 用 generation 来 fencing 这种提交，这是对的；但fencing 只能挡住提交，挡不住已经发生的副作用。扣款在 t1 已经完成，t5 的拒绝只是让这个事实暴露得更晚。

max.poll.interval.ms 是这条时间线上最容易被忽略的参数。它约束的是「两次 poll 之间的最长时间」，不是「一条记录的处理时间」。如果一批 500 条记录、每条 100 ms，一批就要 50 秒；默认 5 分钟的间隔看似够，但下游一抖动就会超。超时后 consumer 被踢出组，分区转走，处理到一半的批次变成重复。调小 max.poll.records 比调大 max.poll.interval.ms 更安全：前者让每批更短，后者只是把炸弹推迟。

再平衡本身也在进化。早期的 eager 协议会让所有消费者先放弃全部分区，再重新分配，整个组停摆几秒到几分钟；cooperative sticky 只迁移需要移动的分区，其他分区继续消费。静态成员（group.instance.id）让滚动重启的消费者在 session.timeout.ms 内保留成员身份，避免每次重启都触发一轮 rebalance。这些机制减少的是停摆时间，不是重复本身；重复仍然要靠幂等处理。

正确的消费循环（示意）

while running:
    records = consumer.poll(500ms)
    for r in records:
        handle(r)                    # 必须幂等
    consumer.commitSync()            # 处理完才提交

on_partitions_revoked(parts):
    consumer.commitSync()            # 尽力而为；失败就交给下游去重
    drain_local_queue()              # 停止接收新活，处理完手上的

这段伪代码有两个细节：提交在处理之后，以及 revoke 时先提交再清空本地队列。第二点经常被漏掉：分区被收走时，本地队列里还有没处理完的记录，如果直接退出，这些记录会在新消费者那里重放；如果先处理完再提交，重放的是「已提交 offset 之后」的部分，重复更少。

如果输入和输出都在 Kafka，可以用事务把「写输出」和「提交 offset」合成一个原子操作：sendOffsetsToTransaction 把 consumer offset 写进 producer 的事务，commitTransaction 同时生效。这是 7.5 要讲的内容。但即使这样，外部数据库或 HTTP 副作用仍然不在事务里，还是需要业务幂等。

7.4.1 提交失败的四种形态

「提交失败」不是一个单一事件，它对数据的影响取决于失败发生在哪一步：

前三行是「至少一次」的重复，后两行是「最多一次」的丢失。很多人只盯着 CommitFailedException，却忽略了 revoke 回调里的顺序：如果在 onPartitionsRevoked 里先 commitSync()、再清空本地队列，那些还没处理的记录已经被 offset 越过了，新消费者不会再来一遍。提交的时机比提交的 API 更重要。

异步提交的乱序是另一个隐蔽的丢失源：commitAsync(104) 还在路上，commitAsync(110) 已经成功，broker 只保留较大的 110，104–109 永远不会被重放。修法是维护一个单调递增的「已提交位点」，只在成功回调里推进，失败时不要提交更大的 offset；或者在 rebalance 时用同步提交兜底。

7.5 Kafka 的 ISR、acks、幂等生产者、事务

先回答一个常见疑问：acks=all 为什么还会丢数据？

因为 acks=all 等的是当前 ISR 里的所有副本，不是「配置的 replication factor 个副本」。ISR（in-sync replicas）是「跟得上 leader 的副本集合」，判断标准是副本最后一次成功 fetch 的时间与 leader 的差距（replica.lag.time.max.ms）。如果一个 follower 因为磁盘慢、网络抖动或 GC 落后太多，它会被踢出 ISR；此时 ISR 可能只剩 leader 自己，acks=all 就退化成 acks=1。

min.insync.replicas 是把 acks=all 变成有意义承诺的配套参数。它规定「至少要有多少个 ISR 才接受写入」；当 ISR 小于这个值时，producer 收到 NotEnoughReplicas，写入失败。这是用可用性换持久性：副本不足时宁可拒绝写，也不接受可能丢的写。典型配置是 replication factor=3、min.insync.replicas=2、acks=all，容忍一个 broker 故障；如果三个副本在同一个机架或同一个可用区，一次机架掉电仍然可能带走多数派。

还有一个更隐蔽的丢数据入口：unclean leader election。当 leader 挂掉、ISR 里没有可用的副本时，如果允许「非 ISR 副本」当 leader，集群能恢复可用，但会丢掉未同步的记录。unclean.leader.election.enable=false 是大多数业务 topic 的默认选择；把它设成 true，等于接受「可用性优先于已确认写」。

HW（high watermark）是消费者能读到的位置。leader 把记录写入本地日志后，要等 ISR 副本都 fetch 到，才推进 HW；consumer 只能读到 HW 之前的记录。这保证了「消费者读到的数据不会因为 leader 切换而消失」。事务场景下还有 LSO（last stable offset）：read_committed 消费者只能读到 LSO 之前、且事务已提交的记录，未提交事务之后的记录一律不可见。

幂等生产者解决的是另一类重复：producer 发送超时后重试，broker 其实已经写入。开启 enable.idempotence=true 后，producer 启动时从 broker 拿到一个 PID（producer id），并为每个分区维护单调递增的 sequence；broker 按 (PID, partition, sequence) 去重，重复的批次直接丢弃，同时保证分区内的顺序。

它有三个边界：

1. 只在同一个 producer 会话内有效。producer 重启会拿到新 PID，之前的 sequence 记录不再匹配；跨会话的重复仍然要靠业务去重。
2. 只覆盖 Kafka 内部的写入。它不知道你的数据库、HTTP 调用有没有重复。
3. 有配置前提：需要 acks=all、retries>0、max.in.flight.requests.per.connection<=5。在不开启幂等时，max.in.flight>1 加上重试会导致批次乱序；开启幂等后，broker 会拒绝乱序的 sequence，producer 也会保序。

事务把范围再扩大一步。给 producer 配置 transactional.id 后，initTransactions() 会向 broker 注册并 fence 掉同 ID 的旧实例（epoch 递增），然后可以：

producer.initTransactions()
producer.beginTransaction()
producer.send(topicA, record1)
producer.send(topicB, record2)
producer.sendOffsetsToTransaction(offsets, groupMetadata)
producer.commitTransaction()        # 或 abortTransaction()

sendOffsetsToTransaction 把消费者 offset 和输出记录放进同一个事务：要么输出可见且 offset 前进，要么都不可见且 offset 不动。消费者设置 isolation.level=read_committed 后，只能读到已提交事务的记录，未提交的记录被 LSO 挡住。这就是 Kafka 里「读-处理-写」的有效一次。

这张表的最后一行是关键：Kafka 的事务不是分布式事务。它不能把「写数据库」和「发消息」包在一起，也不能让外部支付接口参与提交。它能做的是「Kafka 内部的输出和 offset 原子」，把跨系统的部分留给 outbox 和幂等（7.6、7.10）。

长事务还有一个运行时代价：未提交事务会挡住 LSO，read_committed 消费者读不到它之后的记录，lag 看起来正常但消费停住。事务超时（transaction.timeout.ms）和 transaction.max.timeout.ms 要匹配，否则 broker 会主动 abort，producer 收到 InvalidTxnTimeout。这类问题在 7.8 的 lag 读数里表现为「lag 不涨也不降」，只看 lag 很容易误判。

7.5.1 ISR 为什么会缩水

ISR 的判定是「follower 最近一次成功 fetch 距现在多久」，超过 replica.lag.time.max.ms 就被踢出。能让 follower 停止 fetch 的原因有四类：

注意第三行：心跳正常不代表 ISR 正常。broker 之间有心跳，但 ISR 看的是数据 fetch 的进度；一个 GC 停顿的 follower 可能还在回心跳，却已经几秒没有 fetch，照样会被踢出。所以「broker 活着」和「副本同步」是两个监控项。

Kafka 用时间而不是字节数判断落后，是有原因的。按字节判断时，一条大消息就能让副本「看起来落后很多」，产生误报；按时间判断更接近「这个副本还能不能赶上」。LinkedIn 的公开材料也提到，他们把副本落后的判断从字节改成了时间，避免大消息导致的假告警。监控 ISR 要看时间维度的持续性，不要看某一个瞬间的字节差。

ISR 缩水时，acks=all 的承诺跟着缩水：3 副本变 2 副本还能写，变 1 副本就只等 leader；如果 min.insync.replicas=2，1 副本时 producer 会收到 NotEnoughReplicas，写入失败。这不是 bug，是设计：宁可让写入失败，也不让客户端以为写成功了、实际上只在一个随时会挂的 leader 上。

7.6 Outbox、CDC 与事件溯源

第 6.7 节已经讲过 outbox 解决双写：业务行和消息行在同一个本地事务里提交，relay 负责把消息搬出去。这一节换一个视角：outbox 是一张日志表，relay 是一个生产者，consumer offset 是另一张日志的位置。一条事件从数据库到消费者，至少经过三段日志：数据库的 WAL/binlog、outbox 表、Kafka 的分区日志。每一段都有自己的顺序、保留和重复窗口。

先看表结构。下面是 PostgreSQL 版本，重点是约束：

CREATE TABLE orders (
  order_id      TEXT PRIMARY KEY,
  account_id    TEXT NOT NULL,
  amount_cents  BIGINT NOT NULL,
  status        TEXT NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE outbox (
  id            BIGSERIAL PRIMARY KEY,
  event_id      UUID NOT NULL UNIQUE,
  aggregate_id  TEXT NOT NULL,
  aggregate_seq BIGINT NOT NULL,
  event_type    TEXT NOT NULL,
  payload       JSONB NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  published_at  TIMESTAMPTZ,
  UNIQUE (aggregate_id, aggregate_seq)
);

CREATE INDEX outbox_pending ON outbox (id) WHERE published_at IS NULL;

CREATE TABLE processed_events (
  consumer     TEXT NOT NULL,
  event_id     UUID NOT NULL,
  processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (consumer, event_id)
);

四个约束各有作用：event_id 唯一，保证同一条事件不会被业务重复写入；(aggregate_id, aggregate_seq) 唯一，保证同一个聚合的事件序号不重复，消费端可以用它拒绝旧事件；outbox_pending 是部分索引，只索引未发布的行，relay 扫描不会随着历史增长而变慢；processed_events 的主键是 (consumer, event_id)，不同消费者各自去重，互不干扰。

下面是一个可运行的 SQLite 版本，演示三件事：outbox 与业务同事务、relay 在「发布成功、标记失败」时重发、消费者在「处理成功、提交 offset 失败」后重放但不重复扣款。代码只用标准库，直接 python3 outbox_demo.py 即可运行。

import json, sqlite3, uuid

SCHEMA = """
CREATE TABLE orders (
  order_id TEXT PRIMARY KEY, amount_cents INTEGER NOT NULL, status TEXT NOT NULL
);
CREATE TABLE outbox (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  event_id TEXT NOT NULL UNIQUE,
  aggregate_id TEXT NOT NULL,
  aggregate_seq INTEGER NOT NULL,
  event_type TEXT NOT NULL,
  payload TEXT NOT NULL,
  published_at TEXT,
  UNIQUE (aggregate_id, aggregate_seq)
);
CREATE TABLE processed_events (
  consumer TEXT NOT NULL, event_id TEXT NOT NULL,
  PRIMARY KEY (consumer, event_id)
);
CREATE TABLE balances (
  account_id TEXT PRIMARY KEY, balance_cents INTEGER NOT NULL
);
CREATE TABLE offsets (
  consumer TEXT NOT NULL, topic TEXT NOT NULL, next_offset INTEGER NOT NULL,
  PRIMARY KEY (consumer, topic)
);
"""

def connect():
    conn = sqlite3.connect(":memory:")
    conn.executescript(SCHEMA)
    conn.execute("INSERT INTO balances VALUES ('acct-1', 0)")
    conn.commit()
    return conn

def place_order(conn, order_id, amount_cents):
    event_id = str(uuid.uuid4())
    with conn:                              # 业务行 + outbox 行，同一个事务
        conn.execute("INSERT INTO orders VALUES (?, ?, 'created')",
                     (order_id, amount_cents))
        conn.execute(
            "INSERT INTO outbox (event_id, aggregate_id, aggregate_seq, event_type, payload) "
            "VALUES (?, ?, 1, 'order.created', ?)",
            (event_id, order_id, json.dumps({"order_id": order_id,
                                             "amount_cents": amount_cents})),
        )
    return event_id

def relay(conn, topic, crash_after_publish=False):
    rows = conn.execute(
        "SELECT id, event_id, event_type, payload FROM outbox "
        "WHERE published_at IS NULL ORDER BY id"
    ).fetchall()
    for row in rows:
        topic.append({"event_id": row[1], "type": row[2],
                      "payload": json.loads(row[3])})
        if crash_after_publish:             # 发布成功，标记失败
            raise RuntimeError("relay crashed after publish")
        with conn:
            conn.execute("UPDATE outbox SET published_at = datetime('now') "
                         "WHERE id = ?", (row[0],))
    return len(rows)

def consume(conn, topic, consumer, commit_offset=True):
    row = conn.execute(
        "SELECT next_offset FROM offsets WHERE consumer = ? AND topic = 'orders'",
        (consumer,),
    ).fetchone()
    pos = row[0] if row else 0
    handled = 0
    for event in topic[pos:]:
        with conn:                          # 去重记录 + 副作用，同一个事务
            cur = conn.execute(
                "INSERT INTO processed_events (consumer, event_id) VALUES (?, ?) "
                "ON CONFLICT DO NOTHING", (consumer, event["event_id"]))
            if cur.rowcount == 1:
                conn.execute(
                    "UPDATE balances SET balance_cents = balance_cents + ? "
                    "WHERE account_id = 'acct-1'",
                    (event["payload"]["amount_cents"],))
                handled += 1
            if commit_offset:                # 模拟提交失败时置 False
                conn.execute(
                    "INSERT INTO offsets (consumer, topic, next_offset) "
                    "VALUES (?, 'orders', ?) ON CONFLICT (consumer, topic) "
                    "DO UPDATE SET next_offset = excluded.next_offset",
                    (consumer, pos + 1))
        pos += 1
    return handled

conn = connect()
place_order(conn, "order-1", 100)
topic = []
try:
    relay(conn, topic, crash_after_publish=True)
except RuntimeError as exc:
    print("relay:", exc)
relay(conn, topic)                          # 重试：同一条事件再发一次
print("topic copies:", len(topic))
print("handled, commit failed:", consume(conn, topic, "ledger", commit_offset=False))
print("handled, replay:", consume(conn, topic, "ledger"))
print("balance:", conn.execute("SELECT balance_cents FROM balances").fetchone()[0])

一次运行的输出：

relay: relay crashed after publish
topic copies: 2
handled, commit failed: 1
handled, replay: 0
balance: 100

怎么读：

• topic copies: 2：relay 在发布后崩溃，重试又发了一次。重复发生在生产端，这是 outbox 的已知代价。
• handled, commit failed: 1：第一遍处理成功，扣了 100；但 commit_offset=False 模拟提交失败，offset 没有前进。
• handled, replay: 0：第二遍从 offset 0 重放，processed_events 里已经有这个 event_id，副作用被跳过。重复被消费端挡掉。
• balance: 100：一次订单只扣一次。

这个 demo 没有模拟「relay 标记成功但消息没发出去」——那需要 broker 端配合。真实系统里 relay 的 publish 是至少一次：先发后标记，发成功标记失败会重发；先标记后发，发失败就丢。Outbox 只能保证「业务数据和消息记录一起提交」，不能保证「消息只发一次」。

CDC（change data capture）把「写 outbox 表」换成「读数据库日志」。Debezium 之类的连接器订阅 MySQL binlog / PostgreSQL logical replication，把每一行变更转成事件；outbox event router 可以把 outbox 表的插入直接路由成 topic 消息。它的好处是业务代码不用改，坏处是表结构变成了事件契约：加一列、改类型、删字段，所有下游都要跟着变。

CDC 也不是「没有重复」。全量快照和增量日志之间有一个重叠窗口：快照读到某行时，binlog 可能已经包含这行的更新，连接器需要按位点/GTID 去重；切换、重启、schema 变更都会重放一段。CDC 的语义仍然是至少一次，消费端仍然要去重。

事件溯源（event sourcing）走得最远：它不把事件当成「数据库变更的通知」，而把事件当成唯一的事实来源。状态不直接写，而是由事件折叠（fold）出来：state = fold(events)。订单的当前状态是 created、paid、shipped 这些事件依次应用的结果；要查当前状态，可以重放，也可以定期做快照。

这带来两个好处：完整的审计历史，以及「回到任意时刻」的能力。也带来两个硬问题：

1. schema 演进。事件一旦写入就不能改，字段改名、类型变化、语义修正都要靠版本化和 upcaster（读旧版本、转新版本）。不像数据库可以 ALTER TABLE 回填，事件日志里的旧事件永远在那里。
2. 重放副作用。重放 order.created 时不能再发一次货、再扣一次款。纯投影（读模型）可以重放，副作用必须隔离到单独的、幂等的处理器里。

事件溯源常和 CQRS 一起出现：写侧只追加事件，读侧维护投影。投影的更新必须是幂等的，因为它会被重放；投影的位点也要持久化，否则每次重启都从头折叠。Kafka 的压实 topic 常被用作「状态 changelog」：每个 key 保留最新状态，重启时先读 changelog 再继续消费。压实和事件溯源是互补的，不是同一件事：前者保留每个 key 的最新值，后者保留全部历史。

重放一条历史 topic，最常用的命令是重置消费者组的 offset：

# 先 dry-run，确认要重放的范围
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group ledger-replay-20261019 --topic payments.charged \
  --reset-offsets --to-earliest --dry-run

# 确认后执行；只影响这个组，不影响在线组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group ledger-replay-20261019 --topic payments.charged \
  --reset-offsets --to-earliest --execute

重放的前提是：目标组是新组或已停用的组，在线组不能被重置；消费者是幂等的；重放产生的副作用要么被关闭，要么写到隔离的环境。重放不是「再读一遍」，而是「再执行一遍」；如果把 handle 里的扣款、发货、发短信原样跑一遍，就是一次人为的重复副作用。

7.6.1 outbox 表的增长与清理

outbox 是一张只增不减的表：每次业务写都多一行，relay 发布后只是把 published_at 填上，不会删行。一个月后，relay 的 WHERE published_at IS NULL 还能走部分索引，但表的体积、备份和 vacuum 成本会一直涨。

清理要满足两个约束：已发布的行至少保留到「不会再被重放」的窗口，未发布的行一行都不能删。如果 outbox 同时承担审计职责，已发布的行要归档到冷存储，而不是直接删。

-- 分批删除，避免长事务和锁表
DELETE FROM outbox
WHERE id IN (
  SELECT id FROM outbox
  WHERE published_at IS NOT NULL
    AND published_at < now() - interval '7 days'
  ORDER BY id
  LIMIT 10000
);

这条语句要循环执行到影响行数为 0。配套的索引是 published_at，不是 id：

CREATE INDEX outbox_published_at ON outbox (published_at);

比清理更重要的是监控「未发布的积压」。它和 consumer lag 是两回事：outbox 积压说明 relay 没把消息发出去，下游可能一条都没收到。

SELECT count(*) FILTER (WHERE published_at IS NULL)          AS pending,
       max(now() - created_at) FILTER (WHERE published_at IS NULL) AS oldest_pending
FROM outbox;

oldest_pending 超过「业务可接受的最大发布延迟」就该告警，而不是等到下游发现数据少了。relay 的 lag 是生产端的 lag，consumer 的 lag 是消费端的 lag；两个都要看，只看一个会漏掉一半的链路。

7.7 失败模式：重复投递与乱序

重复和乱序经常被当成两个独立的问题：重复靠去重表，乱序靠分区键。但在真实事故里，它们会叠加成第三种错误：一条旧事件被重投，落在一条更新的事件之后，把状态改回去。去重表只能挡住「同一个 event_id」，挡不住「不同 event_id、但语义更旧」的事件。

看一条订单取消的时间线：

t2 的重复是「处理成功、响应超时」，t4 的重放本身没有错——至少一次语义下它必然发生。错的是 t4 的处理逻辑没有检查「这个订单是否已经取消」，也没有比较 seq。重复是触发器，乱序是放大器：如果 t4 的 e1 带着 seq=1，而消费者已经见过 seq=2，一个简单的 seq <= last_seq 判断就能把它丢掉。

重复对「增量写」和「绝对值写」的破坏方式不同，乱序则相反：

这张表说明：去重和顺序是两个不同的问题，不能互相替代。去重表解决「同一条事件被应用两次」；版本号/序号解决「旧事件覆盖新状态」。两者都有的消费者，才能同时挡住重复和乱序。

消费者遇到失败时，重试策略会直接改变顺序：

「原地阻塞重试」看起来最安全，但它会把一条毒消息变成整个分区的停摆：后面的消息都等着，lag 一直涨，最后触发 rebalance，未提交的批次又被重放。重试 topic 把主流程解放出来，代价是顺序；如果重试 topic 不按原 key 分区，同一订单的 created 和 cancelled 会落到不同分区，乱序更严重。重试 topic 也必须用同一个分区键，否则你只是把乱序从主 topic 搬到了重试 topic。

顺序修复通常有三种手段，按成本从低到高：

1. 版本号/序号：每个聚合维护单调 seq，消费者保存 last_seq；seq <= last_seq 直接跳过，seq == last_seq + 1 应用，出现空洞则缓存或回退重放。缓存要有上限，否则一个永远不来的事件会让内存无限增长。
2. 状态机：cancelled 是终态，迟到的 created 只能记录为异常事件，不能把状态改回 active。状态机比序号更懂业务，但需要为每个聚合类型单独设计。
3. 数据库 CAS：UPDATE orders SET status='cancelled', version=version+1 WHERE id=? AND version=?，版本不匹配就拒绝。这把顺序检查下沉到存储层，适合跨消费者、跨服务的场景。

三种手段都不能替代去重：版本号挡住旧事件，去重表挡住同一事件；一个事件可能既旧又重复。

7.7.1 生产者重试也会乱序

乱序不只发生在消费端。producer 开启重试、但没开幂等时，max.in.flight.requests.per.connection > 1 允许第二个批次先成功，第一个批次的重试后到，日志里的顺序就反了：

producer                          broker 日志
batch1(seq=1) --超时，进入重试-->   []
batch2(seq=2) --成功------------>   [batch2]
batch1(seq=1) --重试成功-------->   [batch2, batch1]   # 顺序反了

同一个 key 的两条记录，如果分别落在 batch1 和 batch2，消费者看到的顺序就是 batch2 → batch1。这不是消费端能修的：日志本身已经反了，所有从头读的消费者都会看到反序。修法只有两个：把 max.in.flight 降到 1（吞吐下降），或者开启幂等生产者（broker 按 (PID, partition, sequence) 拒绝乱序/重复的批次）。

这也解释了为什么 7.5 说幂等生产者同时保证「不重复」和「分区内有序」：它不只是去重，还给每个批次一个单调序号，让 broker 能判断「这个批次来晚了」。如果业务依赖同一 key 的严格顺序，幂等生产者不是可选项，是前提。

7.8 观测：consumer lag 与 rebalance

lag 是流系统里最容易被误读的指标。它是「位置差」，不是速率，也不是延迟。100 万的 lag 在一个每秒 10 万条的 topic 上是 10 秒，在一个每秒 100 条的 topic 上是近 3 小时；只看绝对值，等于没看。

最常用的一条命令是 kafka-consumer-groups.sh --describe：

kafka-consumer-groups.sh --bootstrap-server broker-1:9092 \
  --describe --group billing-ledger

GROUP            TOPIC              PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG    CONSUMER-ID                                       HOST         CLIENT-ID
billing-ledger   payments.charged   0          184220          184220          0      consumer-billing-ledger-1-6f3a...                 /10.0.3.17   consumer-billing-ledger-1
billing-ledger   payments.charged   1          90104           90574           470    consumer-billing-ledger-1-6f3a...                 /10.0.3.17   consumer-billing-ledger-1
billing-ledger   payments.charged   2          120071          120071          0      consumer-billing-ledger-2-9c11...                 /10.0.3.18   consumer-billing-ledger-2
billing-ledger   payments.charged   3          -               152200          -      -                                                 -            -

怎么读：

逐行看这份输出：

• P0 的 lag 是 0，消费者是 ledger-1，健康。
• P1 的 lag 是 470，且 ledger-1 同时拥有 P0/P1。如果 470 在几分钟内不降，说明这个消费者处理不过来，或者卡在某个下游。
• P2 的 lag 是 0，消费者是 ledger-2，健康。
• P3 的 CURRENT-OFFSET 和 LAG 都是 -，CONSUMER-ID 也是 -：这个分区没有提交过 offset，也没有活跃消费者。它可能是新加的分区，也可能是组已经空了。- 不是 0；如果把它当 0，会误判为「没有积压」。

再看组状态：

kafka-consumer-groups.sh --bootstrap-server broker-1:9092 \
  --describe --group billing-ledger --state

GROUP            COORDINATOR (ID)          ASSIGNMENT-STRATEGY  STATE               #MEMBERS
billing-ledger   broker-3:9092 (3)          range                Stable              2

#MEMBERS 是当前成员数。把它和 --members 的分配放在一起看，才能发现「成员在、但分区没分到」的情况：

kafka-consumer-groups.sh --bootstrap-server broker-1:9092 \
  --describe --group billing-ledger --members --verbose

GROUP            CONSUMER-ID     HOST         CLIENT-ID      #PARTITIONS  ASSIGNMENT
billing-ledger   ...             /10.0.3.17   ...            2            payments.charged(0,1)
billing-ledger   ...             /10.0.3.18   ...            2            payments.charged(2)

这里 ledger-2 只分到 P2，P3 没有出现在它的 ASSIGNMENT 里；结合上一条命令，P3 没有 owner。这不是「分配不均」，而是「P3 没人要」——可能是消费者只订阅了部分分区，或者 rebalance 还没完成。--members 的输出比 lag 更能暴露「谁在干活、谁在闲着」。

最后看分区副本：

kafka-topics.sh --bootstrap-server broker-1:9092 \
  --describe --topic payments.charged

Topic: payments.charged  PartitionCount: 4  ReplicationFactor: 3
  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3
  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1,2
  Partition: 3  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3

P1 的 Isr 只有 2,3，比 Replicas 少一个：有一个副本落后了。如果这个 topic 配置了 min.insync.replicas=2，P1 还能写；如果配置是 3，P1 的写入会被拒绝。Isr 变短是 acks=all 退化的前兆，比 lag 更值得告警。

把上面的读数归纳成一张判断表：

rebalance 本身也要观测。除了 --state，consumer 日志里的 Revoke previously assigned partitions、(Re-)joining group、Coordinator moved 都是信号；统计每小时的出现次数，比事后看 lag 更能定位「为什么老是抖」。JMX 里的 assigned-partitions、commit-latency-avg、records-lag-max 可以配合看，但先看 --describe 的分区级 lag，它比聚合指标更能指出是哪个分区、哪个消费者。

7.8.1 把 lag 换算成时间

lag 只有换算成「追上要多久」才有决策价值。假设当前 lag 是 L，消费速率是 c 条/秒，生产速率是 p 条/秒，那么追平时间近似为：

$$T_{\text{catch-up}} = \frac{L}{c - p}, \quad c > p$$

如果 c <= p，分母为零或负数，lag 永远不会自己降下来——这时加消费者、限流生产、或者降级下游，必须做其中一个，否则积压只会越来越大。

这张表说明两件事。第一，同样的 10 万 lag，追平时间可以从 20 秒到无穷，只看 lag 会得出完全相反的结论。第二，消费速率和生产速率的差比绝对值更重要；c 只比 p 大 10% 时，任何一点抖动都会让差变成负数。

聚合 lag 还会掩盖单分区故障。一个 12 分区的 topic，11 个分区 lag 为 0、1 个分区 lag 为 100 万，总 lag 是 100 万；如果只看总和，会以为「整体还行」，实际上那个分区可能已经卡死。告警要同时看 max(partition_lag) 和「lag 超过阈值的分区数」，而不是只看 sum(lag)。

7.9 真实案例：Kafka 在 LinkedIn 的规模与运维

2019 年 10 月 8 日，LinkedIn 工程博客发表了 Jon Lee 和 Wesley Wu 的《How LinkedIn customizes Apache Kafka for 7 trillion messages per day》。公开的数字是：100 多个 Kafka 集群、4,000 多个 broker、100,000 多个 topic、700 万个分区，每天处理超过 7 万亿条消息；一些大集群超过 140 个 broker，单个集群托管 100 万个 replica。

这些数字本身不是重点，重点是它们带来的约束。把公开材料里的做法按「约束 → 机制」拆开：

表中最后一行值得展开。LinkedIn 的 Kafka 分支不是 fork，而是在 Apache Kafka 的 release branch 上打补丁，补丁尽量 upstream-first；每个内部版本发布前要过一轮认证，认证项包括 rebalance、部署、rolling bounce、稳定性和 downgrade。这不是「大厂才需要」的流程，而是「Kafka 的变更会影响分区分配」这个事实的后果：一次 broker 重启、一次分区迁移、一次版本升级，都可能触发 rebalance，而 rebalance 会重放未提交的批次。把 rebalance 当成发布验证的一等公民，比事后查 lag 更划算。

从公开材料里能提炼出四条可迁移的做法，和规模无关：

1. 分区预算是容量决策，不是创建 topic 时的随手填。分区数决定并行度，也决定 controller 元数据、rebalance 时长和恢复时间；LinkedIn 的 700 万分区需要定制 controller，普通团队即使只有几千个分区，也应该把「分区总数」当成一个要监控的配额。
2. 保留和复制是按 workload 定的。活动追踪、指标、数据库 CDC 的保留和副本要求不同，混在一个集群里只能取最严格的那个，成本会失控；按用途隔离集群，才能让每个集群的 retention/replication 匹配它的 SLO。
3. rebalance 要自动化，不要靠 SRE 手工 reassign。broker 故障是常态，Cruise Control 这类工具持续看磁盘、网络、CPU，自动迁移分区；没有自动化，集群越大，人工操作越慢、越容易出错。
4. lag 要看行为，不要只看阈值。LinkedIn 开源的 Burrow 用滑动窗口看 offset 提交的趋势，把消费者判成 good/degraded/stopped，而不是「lag > 10000 就告警」。阈值告警在流量有季节性的系统里会大量误报。

7.9.1 元数据：700 万分区压垮 controller

Kafka 的 controller 负责维护 topic/partition/replica 的元数据，并把变更下发给 broker。分区数越多，controller 的内存、UpdateMetadataRequest 的体积、broker 启动时拉取元数据的时间都会增长。LinkedIn 的公开材料提到，大集群出现过 controller 内存压力和 controller 故障，甚至级联失败；他们通过复用 UpdateMetadataRequest 对象、减少日志、缩短 broker 启停时间等补丁来缓解。

这说明「分区数」不只是并行度参数，还是元数据预算。一个中等规模的集群，如果有几千个 topic、每个 topic 几十个分区，分区总数很容易到十万级；这时 controller 的堆、元数据下发延迟、broker 启动时间都要纳入容量规划。监控至少要有：

7.9.2 复制、保留与成本

7 万亿条/天是一个吞吐数字，乘以保留期和副本数才是存储成本。一个粗略的算法：

每天字节数 = 平均写入带宽 × 86400
总存储     = 每天字节数 × 保留天数 × 副本数 × 1.3（索引/段开销）

假设平均写入 1 GB/s、保留 7 天、3 副本：1 GB/s × 86400 ≈ 86.4 TB/天，乘 7 天再乘 3 约 1.8 PB。这个数字随保留期线性增长，所以「保留久一点」不是免费的运维选项，而是容量和成本的决策。

LinkedIn 的做法是按 workload 分集群：活动追踪、指标、数据库 CDC 的保留、副本和延迟要求不同，混在一起只能取最严格的那个。保留期的下界是「最大消费者停机时间 + 全量重放时间」，上界是预算；超过下界之后，每多一天都要问「这一天真的会被重放吗」，否则就是在为「可能有用」付钱。副本也是同理：3 副本是常见默认，但冷数据、可重建的派生数据可以用更低的副本数，前提是接受重建时间。

7.9.3 把 rebalance 当成发布对象

LinkedIn 的发布认证包含 rebalance、部署、rolling bounce、稳定性和 downgrade。这个列表值得抄，因为它覆盖的正是「变更如何影响分区分配」：

• rebalance：滚动重启消费者，观察重复处理、lag 峰值、分区分配是否均匀；
• rolling bounce：逐台重启 broker，观察 ISR 是否反复缩水、producer 是否收到 NotEnoughReplicas；
• downgrade：新版本回退到旧版本，元数据和事务状态是否兼容；
• stability：在持续写入下跑一段时间，看 controller、GC、磁盘是否稳定。

大多数团队不需要定制 Kafka 分支，但需要一个「发布前跑一遍 rebalance」的检查项。rebalance 不是异常，是发布的一部分；把它当成异常，就会在每次发版时被同一个重复扣款打中。

7.10 重试导致重复副作用：offset 提交失败后的二次扣款

7.4 的时间线不是假设。把场景换成一个钱包服务：它消费 payments.charged，调用支付服务商（PSP）扣款，然后写本地账本。某次滚动发版，consumer 超过了 max.poll.interval.ms，再平衡发生，offset 提交被拒；新 consumer 从旧 offset 重放，用户被扣了两次。

根因不是「offset 提交失败」——它是必然发生的；根因是幂等键在每次处理时重新生成，以及 charges 表没有以业务意图为唯一键。K1 和 K2 是两个不同的键，PSP 无法知道它们对应同一笔订单；本地两条 charge 记录也让对账看起来像两笔合法扣款。

修复要分四层，缺一层都会留下窗口：

表结构：

CREATE TABLE charges (
  order_id     TEXT PRIMARY KEY,
  event_id     UUID NOT NULL UNIQUE,
  amount_cents BIGINT NOT NULL,
  status       TEXT NOT NULL CHECK (status IN ('pending', 'succeeded', 'failed')),
  idem_key     TEXT NOT NULL UNIQUE,
  psp_ref      TEXT,
  updated_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

order_id 主键是业务意图的粒度：一个订单只允许一条扣款记录。event_id 唯一是消费去重的粒度：同一条事件重放不会插入第二行。idem_key 是传给 PSP 的键，和 order_id 一样稳定；status 让「已经扣了但没记账」和「还没扣」区分开；psp_ref 是对账的锚点。

处理逻辑的关键是「先占位，再调用，失败可查」：

def handle(event, conn, psp):
    order_id = event["payload"]["order_id"]
    key = event["event_id"]                  # 稳定，不随重试变化
    row = conn.execute(
        "INSERT INTO charges (order_id, event_id, amount_cents, status, idem_key) "
        "VALUES (%s, %s, %s, 'pending', %s) "
        "ON CONFLICT (order_id) DO NOTHING RETURNING order_id",
        (order_id, event["event_id"], event["payload"]["amount_cents"], key),
    ).fetchone()
    if row is None:                          # 这个订单已经有一条 charge
        status, ref = conn.execute(
            "SELECT status, psp_ref FROM charges WHERE order_id = %s", (order_id,)
        ).fetchone()
        if status == "succeeded":
            return ref
        return psp.query(key)                # pending：先查，不盲目再扣
    ref = psp.charge(key, order_id, event["payload"]["amount_cents"])
    conn.execute(
        "UPDATE charges SET status = 'succeeded', psp_ref = %s WHERE order_id = %s",
        (ref, order_id),
    )
    return ref

这段代码有两个必须理解的点。第一，INSERT ... ON CONFLICT DO NOTHING 是原子占位：两个并发重放只有一个能插入成功，另一个走查询分支。第二，psp.charge 在数据库事务之外调用；如果 PSP 成功、进程在 UPDATE 前崩溃，记录停在 pending，下次重放会走 psp.query(key)，用同一个键拿回原来的扣款结果，而不是再扣一次。状态机把「未知」变成了可查询的 pending，而不是靠重试猜。

query 返回 not_found 时才允许用同一个键重新 charge；如果 PSP 不支持按幂等键查询，就把 pending 记录交给对账任务，而不是在消费循环里盲目重试。没有查询能力的幂等键，只是一个「可能重复」的键。

对账是最后一道防线。即使前四层都做了，也可能有历史脏数据或代码 bug；每天跑一次「一个订单多笔扣款」的查询：

SELECT order_id, COUNT(*) AS charges, SUM(amount_cents) AS total
FROM psp_charges
WHERE created_at >= now() - interval '1 day'
GROUP BY order_id
HAVING COUNT(*) > 1;

对账发现重复后，退款/冲正本身也要幂等：用 order_id 作为退款键，先查是否已经退过，再调用 PSP。否则「修复重复」会制造新的重复。

7.11 事件设计清单

事件一旦发布，就是一份公开契约。消费者会按字段解析、按类型分支、按 event_id 去重；你改一个字段名，可能让三个团队当天上线失败。设计事件时，先把信封（envelope）定下来，再谈 payload。

{
  "event_id": "01J9Z8K3Q7YV2M6T4R1X5B8N0C",
  "type": "payment.charged",
  "version": 3,
  "occurred_at": "2026-10-19T09:31:02.114Z",
  "producer": "payments-api/2.14.0",
  "aggregate_id": "order-8891",
  "aggregate_seq": 7,
  "partition_key": "order-8891",
  "correlation_id": "trace-4f2a9c...",
  "causation_id": "01J9Z8K3Q7YV2M6T4R1X5B8N0B",
  "payload": {
    "order_id": "order-8891",
    "amount_cents": 12800,
    "currency": "CNY"
  }
}

信封里的字段各有职责，不能互相替代：

这张表里最容易被跳过的是第 7、8、12 项。没有 aggregate_seq，消费端只能用 occurred_at 猜顺序，而时钟不可靠；没有稳定的 partition_key，同一聚合的事件会散到不同分区；没有保留/重放设计，消费者停机一周后只能面对 OffsetOutOfRange。事件设计不是 JSON 字段设计，而是「身份、顺序、重放、演进」四件事的契约。

事件、命令、快照经常被混用，语义不同：

把命令当事件发，会让下游以为「请求一定成功」；把事件当命令发，会让下游重复执行。事件用过去式，命令用祈使句，命名是最便宜的类型检查。

分区键的规则可以压缩成四句话：

1. 同一聚合的所有事件用同一个键，通常是 aggregate_id；
2. 键一旦发布就不要改；要改就双写新旧 topic，做受控迁移；
3. 有序流程不要用 null key；null key 的 sticky 分区没有 key 级顺序；
4. 键的基数要足够高，避免把大部分流量压到一个分区；热 key 要单独设计。

schema 演进要按「旧消费者读新事件」和「新消费者读旧事件」两个方向检查：

保留和重放要一起设计。保留期至少覆盖「消费者最长停机时间 + 全量重放时间」；超过这个窗口的数据要归档到对象存储，并保留可重放的格式。压实 topic 适合保存「每个 key 的当前状态」，不适合保存「所有历史事件」；墓碑的保留时间（delete.retention.ms）要大于消费者的最大停机时间，否则删除事件会先消失，消费者永远不知道某个 key 被删了。

7.12 本章陷阱清单

7.13 本章出口检查

母本对应

这一章没有一一对应的母本章节：谢希仁和张彦飞讲的是网络分层与内核路径，不覆盖队列/日志语义、消费者组、outbox 和事件溯源。本书自编的部分集中在「机制 → 失败窗口 → 观测命令 → 修复」这条链上；Kafka 的 ISR、事务、幂等生产者等事实来自论文与官方文档，代码和事故场景是按工程实践重写的，不复制文档长段，也不把 LinkedIn 的规模当成默认配置。

下一站：消息把「一次状态变化」变成了可重放的流，但还有一类数据不在你的数据库里：对象、缓存和 CDN。进入 8 对象存储、缓存与 CDN，看不可变对象、条件写、缓存失效和回源风暴如何接住这条流的下游。