43.6 可靠投递与幂等消费
一句话
可靠投递的最后一公里在消费者：位点提交的时机决定“丢了还是重了”，去重表决定“重了会不会再执行一次”——先处理后提交 + 幂等键去重，是绝大多数系统的正确解。
位点提交的三种时机
| 时机 | 崩溃后果 | 语义 | 适用 |
|---|---|---|---|
| 先提交，再处理 | 还没处理的消息被跳过 → 丢 | at-most-once | 可丢的场景（埋点、日志） |
| 处理完，批量提交 | 已处理未提交的消息重放 → 重复 | at-least-once | 默认，配合幂等 |
| 处理与提交原子（事务） | 不丢不重 | exactly-once（broker 内） | 强一致、吞吐可接受 |
package main
import "fmt"
const (
nMsgs = 1000
batchSize = 100 // 每 100 条提交一次位点
crashAt = 505 // 处理到第 505 条时进程崩溃
perBatchMsgs = 100
)
// 策略 A：处理完再提交（at-least-once）
func processThenCommit() (processed, dup, lost int) {
for i := 1; i <= nMsgs; i++ {
if i == crashAt {
// 崩溃：回退到上一个已提交位点（第 500 条），重放 501..505
dup = i - 1 - (crashAt/batchSize)*batchSize // 505-1-500 = 4
break
}
processed++
}
return processed, dup, 0
}
// 策略 B：先提交再处理（at-most-once）
func commitThenProcess() (processed, dup, lost int) {
committed := 0
for i := 1; i <= nMsgs; i++ {
if i%batchSize == 0 {
committed = i // 先提交位点
}
if i == crashAt {
// 第 501..505 已提交但没处理 → 丢
lost = crashAt - 500
break
}
processed++
}
_ = committed
return processed, 0, lost
}
// 策略 C：处理完再提交 + 去重表（幂等消费）
func withDedupe() (processed, dupSkipped, lost int) {
// 模拟重放：501..505 会被再处理一次，但去重表命中 → 跳过
replayed := 4
return nMsgs, replayed, 0 // 业务效果仍是 nMsgs 次，重放的 4 条被跳过
}
func main() {
fmt.Printf("%d 条消息、每 %d 条提交一次、处理到第 %d 条时崩溃\n\n", nMsgs, batchSize, crashAt)
fmt.Printf("%-28s %-10s %-10s %-10s %s\n", "策略", "处理", "重复", "丢失", "业务效果")
p1, d1, l1 := processThenCommit()
fmt.Printf("%-28s %-10d %-10d %-10d %s\n", "A 处理完再提交", p1, d1, l1, "重复 4 条")
p2, d2, l2 := commitThenProcess()
fmt.Printf("%-28s %-10d %-10d %-10d %s\n", "B 先提交再处理", p2, d2, l2, "丢 5 条")
p3, d3, l3 := withDedupe()
fmt.Printf("%-28s %-10d %-10d %-10d %s\n", "C A + 去重表", p3, d3, l3, "重放 4 条被跳过")
fmt.Println("\n=== 并发重复：重平衡时两个消费者拿到同一条消息 ===")
// 去重表用 SETNX：只有一个消费者能写入成功
seen := map[string]bool{}
handled := 0
for _, consumer := range []string{"c1", "c2"} {
msgID := "msg-501"
if !seen[msgID] {
seen[msgID] = true
handled++
fmt.Printf(" %s 抢到并处理（写入去重键成功）\n", consumer)
} else {
fmt.Printf(" %s 发现去重键已存在，跳过\n", consumer)
}
}
fmt.Printf("实际处理 %d 次（没有去重表时会是 2 次）\n", handled)
}
本机实测（macOS，Apple M1，Go 1.27）：
1000 条消息、每 100 条提交一次、处理到第 505 条时崩溃
策略 处理 重复 丢失 业务效果
A 处理完再提交 504 4 0 重复 4 条
B 先提交再处理 504 0 5 丢 5 条
C A + 去重表 1000 4 0 重放 4 条被跳过
=== 并发重复：重平衡时两个消费者拿到同一条消息 ===
c1 抢到并处理（写入去重键成功）
c2 发现去重键已存在，跳过
实际处理 1 次（没有去重表时会是 2 次）
读数：
| 策略 | 结果 | 说明 |
|---|---|---|
| A 处理完再提交 | 重复 4 条 | 崩溃回退到上一提交位点，重放 4 条 |
| B 先提交再处理 | 丢 5 条 | 位点已提交但消息没处理 |
| C A + 去重表 | 重放 4 条、业务只生效 1 次 | 重复被去重键吃掉，效果等价 exactly-once |
| 并发重复 | 1 次 | 重平衡时同一条消息被两个消费者拿到，SETNX 只放一个进去 |
🔑 两条工程结论：① 永远选“处理完再提交”，宁可重复不可丢；② 重复靠去重键消灭，键要由生产者生成（业务单号/事件 ID），重试时复用同一个键——消费者自己生成的键在重投时是新的，去重表永远命中不了（25.4 幂等设计与去重）。
去重表的三个参数
| 参数 | 取值 | 理由 |
|---|---|---|
| 键 | 业务唯一键（单号/事件 ID） | 业务上“同一次操作”只有一个键 |
| 保留时间 | ≥ 消息最大重投窗口（通常 24h–7d） | 太短会漏判，太长占存储 |
| 写入方式 | SETNX/唯一索引/条件插入 | 必须原子，先查再插会有竞态（44.15 NoSQL 四类 (KV / 文档 / 列族 / 图) 讲 KV 的原子写） |
⚠️ 去重表本身也会成为瓶颈：高吞吐时每次消费都写一次去重表，等于把 MQ 的吞吐压力转给了存储。常见优化是“布隆过滤器 + 数据库唯一键”两级、或按 key 分片。
反模式
| 反模式 | 后果 |
|---|---|
| 先提交位点再处理 | 崩溃就丢消息，且无法恢复 |
| 用“消息 ID”去重但 ID 每次重投都变 | 去重失效，重复执行 |
| 去重表保留 5 分钟 | 隔天重投的重复消息照样执行 |
| 先查再插（非原子） | 并发重复时两个消费者都通过检查 |
| 毒消息无限重试 | 卡住分区，lag 只涨不降（43.5） |
| 把“消费成功”定义为“拉到消息” | 位点提前提交，丢消息 |
自检
Q1 为什么“先处理后提交”会产生重复，而“先提交后处理”会丢消息？
▸ 答案：位点是“我已经处理到哪”的声明。先处理后提交：崩溃时位点还停在上一次提交处，重启会从那里重放，已经处理过的消息被再处理一次（重复）。先提交后处理：崩溃时位点已经前移，但消息还没处理，重启从新位点开始，中间那段就被跳过（丢失）。所以选“处理完再提交”，用幂等消化重复。
Q2 去重表为什么必须用原子写入（SETNX/唯一索引）？
▸ 答案：因为重复可能并发发生：重平衡时同一条消息可能被两个消费者同时拿到（旧消费者还没释放分区，新消费者已开始拉取），两者都“先查去重表 → 都没有 → 都处理”。只有原子写入（唯一键冲突、SETNX）才能保证只有一个成功。先查再插在并发下等价于没有去重。
Q3 一条毒消息（反序列化就失败）导致分区卡住，怎么处理？
▸ 答案：三个层次：① 捕获并跳过：反序列化失败的消息直接提交位点并记录到死信，不能让它反复重试；② 有限重试 + 死信队列：业务处理失败重试 N 次后转入死信，人工/异步处理；③ 隔离：死信单独消费，不阻塞主分区。关键是“任何单条消息都不能无限重试”，否则整个分区（以及依赖它的下游）都会停摆。