“至少一次”不是实现缺陷
4.1 的 Dispatcher 在处理器失败时把事件留在 pending，下一轮再试；即使投递成功，若进程在标记前崩溃，消息也会再投一次。代理侧能做到的语义只有至少一次或最多一次，“恰好一次投递”在跨系统边界上不成立。消息队列自身的重投、分区和确认机制已在 消息队列深度解析
中讲过，本节只回答一个问题：重复到达时，业务怎样只生效一次。
先给结论：业务层追求的不是“消息只来一次”，而是“副作用只发生一次”。这需要去重键、处理状态和终态记录三者配合；只用一个 seen 集合，无法区分“正在处理”和“已经处理完”，并发重复与崩溃恢复都会漏。
三种重复要分开处理
第一种是代理重投：同一条 message_id 在短时间内到达多次。去重表命中 DONE 就可以直接确认，不执行业务。
第二种是并发重复：两个消费者实例同时拿到同一条消息。此时第一条还在处理中，第二条不能执行，也不能简单丢弃后确认；正确做法是返回“处理中”，让代理稍后重投，或由调用方延迟确认。
第三种是处理中崩溃：消费者拿到租约、开始执行业务，进程在写终态前挂掉。去重表里留下 PROCESSING，租约到期后必须允许另一个实例接管，否则这条消息永久卡死。
这三种情况对应三个状态转换，不能靠一个布尔值表达。Record 因此需要 State、LeaseUntil、Attempts 和 UpdatedAt。
消费循环与去重存储
Consume 先调用 Begin 申请处理权：新消息返回 claimed=true；DONE 返回重复；PROCESSING 且租约未过期返回 ErrInFlight；租约过期则接管并递增尝试次数。业务成功后写 DONE，失败时故意不清租约，让失败与崩溃走同一条恢复路径。
// file: internal/consumer/consumer.go
// tier: L1
// verify: go test -race -v ./internal/consumer/
package consumer
import (
"context"
"errors"
"fmt"
"sync"
"time"
)
var ErrInFlight = errors.New("message is being processed by another consumer")
type State string
const (
StateProcessing State = "PROCESSING"
StateDone State = "DONE"
)
type Result string
const (
ResultApplied Result = "APPLIED"
ResultDuplicate Result = "DUPLICATE"
ResultInFlight Result = "IN_FLIGHT"
)
type Message struct {
ID string
Topic string
}
// Record 是去重表的一行：message_id + 处理状态 + 租约。
// 只有“已处理/未处理”两态无法区分“正在处理”和“可以抢占”。
type Record struct {
State State
Attempts int
LeaseUntil time.Time
UpdatedAt time.Time
}
type Handler func(context.Context, Message) error
type Store interface {
Begin(ctx context.Context, consumer, messageID string, now time.Time, lease time.Duration) (Record, bool, error)
Complete(ctx context.Context, consumer, messageID string, now time.Time) error
Prune(ctx context.Context, before time.Time) (int, error)
}
// Consumer 实现至少一次消费：代理可以重复投递，业务效果由去重记录收敛到一次。
type Consumer struct {
Name string
Store Store
Lease time.Duration
Now func() time.Time
Handle Handler
}
func (c *Consumer) Consume(ctx context.Context, m Message) (Result, error) {
lease := c.Lease
if lease <= 0 {
lease = 30 * time.Second
}
now := time.Now()
if c.Now != nil {
now = c.Now()
}
rec, claimed, err := c.Store.Begin(ctx, c.Name, m.ID, now, lease)
if err != nil {
return "", err
}
if !claimed {
if rec.State == StateDone {
return ResultDuplicate, nil
}
return ResultInFlight, ErrInFlight
}
if err := c.Handle(ctx, m); err != nil {
// 不主动清租约：失败与崩溃走同一条恢复路径，等租约到期后重试。
return "", err
}
if err := c.Store.Complete(ctx, c.Name, m.ID, now); err != nil {
return "", err
}
return ResultApplied, nil
}
// MemoryStore 是单实例演示。生产实现把 (consumer, message_id) 放进唯一索引，
// 并与业务写入共享同一事务。
type MemoryStore struct {
mu sync.Mutex
records map[string]Record
}
func NewMemoryStore() *MemoryStore {
return &MemoryStore{records: map[string]Record{}}
}
func key(consumer, messageID string) string { return consumer + "|" + messageID }
func (s *MemoryStore) Begin(ctx context.Context, consumer, messageID string, now time.Time, lease time.Duration) (Record, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
k := key(consumer, messageID)
rec, ok := s.records[k]
if !ok {
rec = Record{State: StateProcessing, Attempts: 1, LeaseUntil: now.Add(lease), UpdatedAt: now}
s.records[k] = rec
return rec, true, nil
}
if rec.State == StateDone {
return rec, false, nil
}
if now.Before(rec.LeaseUntil) {
// 另一个消费者仍持有租约，不能并发执行业务处理。
return rec, false, nil
}
rec.Attempts++
rec.LeaseUntil = now.Add(lease)
rec.UpdatedAt = now
s.records[k] = rec
return rec, true, nil
}
func (s *MemoryStore) Complete(ctx context.Context, consumer, messageID string, now time.Time) error {
s.mu.Lock()
defer s.mu.Unlock()
k := key(consumer, messageID)
rec, ok := s.records[k]
if !ok {
return fmt.Errorf("complete unknown message %s", messageID)
}
rec.State = StateDone
rec.LeaseUntil = time.Time{}
rec.UpdatedAt = now
s.records[k] = rec
return nil
}
func (s *MemoryStore) Prune(ctx context.Context, before time.Time) (int, error) {
s.mu.Lock()
defer s.mu.Unlock()
removed := 0
for k, rec := range s.records {
if rec.State == StateDone && rec.UpdatedAt.Before(before) {
delete(s.records, k)
removed++
}
}
return removed, nil
}
这里有两个容易被忽略的边界。第一，Begin 与 Complete 在内存实现里是两次独立加锁；生产实现必须把“业务写入 + 去重记录完成”放进同一个事务，否则业务提交成功、状态写失败时仍会重复执行。第二，失败不解租约意味着重试要等租约到期；如果业务需要快速重试，应把失败原因和下次可见时间写入记录，而不是在 Consume 里忙等。
测试覆盖三类重复
测试用固定时钟推进租约：重复投递验证 APPLIED → DUPLICATE → DUPLICATE；并发测试让 6 个 goroutine 抢同一条消息，只允许一个 APPLIED；崩溃测试直接调用 Begin 后不写终态，验证租约内不可抢占、过期后可接管。
// file: internal/consumer/consumer_test.go
// tier: L1
// verify: go test -race -v ./internal/consumer/
package consumer
import (
"context"
"errors"
"sync"
"sync/atomic"
"testing"
"time"
)
var base = time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC)
func fixedConsumer(store Store, calls *atomic.Int32) *Consumer {
var tick atomic.Int64
return &Consumer{
Name: "notifier", Store: store, Lease: 30 * time.Second,
Now: func() time.Time { return base.Add(time.Duration(tick.Add(1)) * time.Second) },
Handle: func(ctx context.Context, m Message) error {
calls.Add(1)
return nil
},
}
}
func TestDuplicateDeliveryAppliesOnce(t *testing.T) {
store := NewMemoryStore()
var calls atomic.Int32
c := fixedConsumer(store, &calls)
want := []Result{ResultApplied, ResultDuplicate, ResultDuplicate}
for i, expected := range want {
got, err := c.Consume(context.Background(), Message{ID: "m-1", Topic: "order.placed"})
if err != nil || got != expected {
t.Fatalf("delivery %d: got=%s want=%s err=%v", i, got, expected, err)
}
}
if calls.Load() != 1 {
t.Fatalf("handler calls=%d want 1", calls.Load())
}
}
func TestConcurrentDuplicatesApplyOnce(t *testing.T) {
store := NewMemoryStore()
var calls atomic.Int32
c := &Consumer{
Name: "notifier", Store: store, Lease: time.Second,
Handle: func(ctx context.Context, m Message) error {
calls.Add(1)
time.Sleep(20 * time.Millisecond)
return nil
},
}
const workers = 6
results := make([]Result, workers)
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
r, err := c.Consume(context.Background(), Message{ID: "m-concurrent"})
if err != nil && !errors.Is(err, ErrInFlight) {
t.Errorf("worker %d: %v", i, err)
}
results[i] = r
}(i)
}
wg.Wait()
applied := 0
for _, r := range results {
if r == ResultApplied {
applied++
}
}
if applied != 1 || calls.Load() != 1 {
t.Fatalf("applied=%d calls=%d results=%v", applied, calls.Load(), results)
}
}
func TestHandlerFailureIsRetried(t *testing.T) {
store := NewMemoryStore()
var calls atomic.Int32
var tick atomic.Int64
c := &Consumer{Name: "notifier", Store: store, Lease: time.Second}
c.Now = func() time.Time { return base.Add(time.Duration(tick.Add(1)) * time.Second) }
c.Handle = func(ctx context.Context, m Message) error {
if calls.Add(1) <= 2 {
return errors.New("provider timeout")
}
return nil
}
ctx := context.Background()
for i := 0; i < 2; i++ {
if _, err := c.Consume(ctx, Message{ID: "m-retry"}); err == nil {
t.Fatalf("attempt %d: want error", i+1)
}
}
if got, err := c.Consume(ctx, Message{ID: "m-retry"}); err != nil || got != ResultApplied {
t.Fatalf("third attempt: result=%s err=%v", got, err)
}
if calls.Load() != 3 {
t.Fatalf("calls=%d want 3", calls.Load())
}
}
func TestCrashBeforeAckLeasesUntilExpiry(t *testing.T) {
store := NewMemoryStore()
ctx := context.Background()
if _, claimed, err := store.Begin(ctx, "notifier", "m-crash", base, time.Second); err != nil || !claimed {
t.Fatalf("first claim: claimed=%v err=%v", claimed, err)
}
// 模拟处理中崩溃：没有 Complete，也没有 Fail，租约留在表里。
if rec, claimed, err := store.Begin(ctx, "notifier", "m-crash", base.Add(500*time.Millisecond), time.Second); err != nil || claimed {
t.Fatalf("within lease: claimed=%v rec=%+v err=%v", claimed, rec, err)
}
rec, claimed, err := store.Begin(ctx, "notifier", "m-crash", base.Add(2*time.Second), time.Second)
if err != nil || !claimed || rec.Attempts != 2 || rec.State != StateProcessing {
t.Fatalf("after lease: claimed=%v rec=%+v err=%v", claimed, rec, err)
}
if err := store.Complete(ctx, "notifier", "m-crash", base.Add(2*time.Second)); err != nil {
t.Fatal(err)
}
if _, claimed, _ := store.Begin(ctx, "notifier", "m-crash", base.Add(3*time.Second), time.Second); claimed {
t.Fatal("completed message must not be claimed again")
}
}
func TestPruneDefinesDedupWindow(t *testing.T) {
store := NewMemoryStore()
var calls atomic.Int32
c := fixedConsumer(store, &calls)
ctx := context.Background()
if got, err := c.Consume(ctx, Message{ID: "m-window"}); err != nil || got != ResultApplied {
t.Fatalf("first: result=%s err=%v", got, err)
}
if removed, err := store.Prune(ctx, base.Add(time.Hour)); err != nil || removed != 1 {
t.Fatalf("prune removed=%d err=%v", removed, err)
}
// 去重记录被清理后，同一个 message_id 会再次执行业务。
if got, err := c.Consume(ctx, Message{ID: "m-window"}); err != nil || got != ResultApplied {
t.Fatalf("after prune: result=%s err=%v", got, err)
}
if calls.Load() != 2 {
t.Fatalf("calls=%d want 2", calls.Load())
}
}
真实运行结果
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），命令 go test -race -v ./... -count=1，原始输出保存在 .local/reports/be-arch/be-arch-perf-04-2-delivery-semantics-consumer-idempotency-race.txt：
=== RUN TestDuplicateDeliveryAppliesOnce
--- PASS: TestDuplicateDeliveryAppliesOnce (0.00s)
=== RUN TestConcurrentDuplicatesApplyOnce
--- PASS: TestConcurrentDuplicatesApplyOnce (0.02s)
=== RUN TestHandlerFailureIsRetried
--- PASS: TestHandlerFailureIsRetried (0.00s)
=== RUN TestCrashBeforeAckLeasesUntilExpiry
--- PASS: TestCrashBeforeAckLeasesUntilExpiry (0.00s)
=== RUN TestPruneDefinesDedupWindow
--- PASS: TestPruneDefinesDedupWindow (0.00s)
PASS
ok archlab/internal/consumer 1.551s
-race 没有报告数据竞争；并发测试里 6 个 goroutine 只有 1 个返回 APPLIED，其余返回 DUPLICATE 或 IN_FLIGHT。这是本机单次运行，不能据此推断线上吞吐。
“恰好一次”在业务层的真实含义
“恰好一次”通常指恰好一次业务效果，而不是恰好一次传输。它成立需要三个条件同时满足：去重键稳定且由生产端生成；去重记录与业务副作用在同一个事务边界内；重复窗口内的消息都能查到去重记录。缺任何一个，就只能退化为“至少一次 + 业务幂等”。
以通知为例，发短信是不可回滚的外部副作用。即使去重表显示 DONE，也可能出现“短信已发出、状态写失败”的窗口；此时重试会再发一次。要彻底消除，需要供应商支持幂等键，或把“已发送”作为业务记录先落库再调用供应商，并接受极小概率的重复。工程上更现实的目标是：重复率可观测、重复可补偿、用户不会被重复扣款。
去重窗口的边界
去重记录不能无限保留。arch-lab 的口径是按事件时间保留 7 到 30 天，并且保留期必须大于代理的最长重投窗口；否则代理在第 8 天重投一条旧消息，去重表已经清空，业务会再执行一次。
Prune 的测试故意展示了这个边界：清理掉 DONE 后，同一个 message_id 会重新返回 APPLIED。这不是 bug，而是窗口设计的结果。窗口之外只能靠业务幂等兜底，例如通知按 order_id + template 覆盖写、支付按幂等键查询供应商结果。
还有一个跨库边界：去重表与业务表不同库时，无法共享事务。此时要么用业务唯一键做最终约束，要么接受“业务成功但状态未写”的重复，并通过对账修正。不要把跨库的两阶段提交当成默认方案。
常见错误与边界
第一，只记录“处理过”的 message_id，不记录处理中状态。并发重复会同时通过检查，业务执行两次。
第二，先执行业务、后写去重记录。进程在两步之间崩溃，重投会再次执行；顺序必须是先占位、再执行、最后写终态。
第三，去重窗口短于代理重投窗口。窗口外的重复没有任何拦截，等于把正确性交给了运气。
第四，把去重键交给消费者生成。同一事件被不同实例消费时键不一致，去重表形同虚设；键必须来自生产端的事件 ID。
小结
至少一次投递无法消除重复，能消除的是重复带来的业务效果。arch-lab 用 message_id + PROCESSING/DONE + 租约 把重复投递、并发重复和崩溃重试收敛到一次业务执行，并用 Prune 明确了去重窗口的边界。下一篇继续处理重复之外的运维问题：积压如何度量、顺序如何保证、毒消息如何进入 DLQ，以及定时对账如何兜住最终一致性。
阅读导航：上一节：4.1 同步改异步的收益与代价 · 下一节：4.3 积压、顺序、死信与定时对账补偿 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。