消费链路要同时回答三个问题
4.2 解决了“同一条消息重复到达”的问题，但生产环境还会问：消息排了多久、同一订单的事件会不会乱序、反复失败的消息会不会堵住整个分区。这三个问题分别对应积压年龄、分区顺序和死信治理，不能靠加消费者或加重试次数蒙混过去。本节在 arch-lab 上实现一条最小的订单事件流水线：按 order_id 分区、按 seq 顺序执行、超过重试上限进 DLQ、修复后重放，并用定时对账补发缺失事件。异步化与背压的通用背景见 异步化与背压控制
，这里只写可运行的部分。
按 order_id 分区，顺序只在分区内成立
全局顺序既昂贵又没必要。订单只要保证同一笔订单的事件按序处理，不同订单之间可以并行；因此分区键必须是 order_id，不能是随机数，也不能是消息 ID。Partition 用 FNV-1a 把同一个订单稳定映射到同一个分区。Deliver 收到 seq 大于当前预期的事件时先放进 pending，不执行；只有当 expected+1 到达后，才按连续序列依次调用处理器。重复或过旧的 seq 直接忽略。这样，网络乱序不会让“已支付”先于“已创建”执行。
积压年龄比积压条数更接近用户感受
积压 10 万条如果是 10 秒内产生的，和积压 1 万条但最老一条等了 20 分钟，对用户的影响完全不同。BacklogAge 取所有未处理事件中最老的 EnqueuedAt，用当前时间相减。arch-lab 的口径是：30 秒进入预警，2 分钟触发值班呼叫；这两个阈值按订单链路的用户可接受等待设定，不是通用常数。
重试上限、DLQ 与重放
毒消息不能让分区永久卡住。事件失败时 attempts 递增，达到 maxAttempts 后写入 DLQ，并把 expected 推进到下一条，后续事件可以继续。DLQ 保留原事件、尝试次数和最后一次错误；修复处理器后调用 ReplayDLQ，成功的事件直接记为已处理，不再塞回有序流，避免旧 seq 打乱当前位点。
定时对账只补事件，不改状态
对账解决的是“状态已经到了、事件却缺失”的窗口。Reconcile 在时间窗口内扫描订单：状态有预期事件（如 PAID 对应 payment.captured），但事件表查不到，就发出补偿事件。它跳过 grace 内刚更新的订单，避免和正常投递打架；它不修改订单状态，避免对账任务成为第二个写 owner。
// file: internal/pipeline/pipeline.go
// tier: L1
// verify: go test ./internal/pipeline/
package pipeline
import (
"context"
"fmt"
"hash/fnv"
"sync"
"time"
)
type Event struct {
ID string
OrderID string
Seq int64
EnqueuedAt time.Time
}
type Handler func(context.Context, Event) error
type DeadLetter struct {
Event Event
Attempts int
Err string
}
type item struct {
ev Event
attempts int
}
type Coordinator struct {
partitions, maxAttempts int
handle Handler
mu sync.Mutex
expected map[int]int64
pending map[int]map[int64]*item
dlq []DeadLetter
}
func New(partitions, maxAttempts int, h Handler) *Coordinator {
return &Coordinator{
partitions: partitions, maxAttempts: maxAttempts, handle: h,
expected: map[int]int64{}, pending: map[int]map[int64]*item{},
}
}
func Partition(orderID string, partitions int) int {
h := fnv.New32a()
_, _ = h.Write([]byte(orderID))
return int(h.Sum32() % uint32(partitions))
}
func (c *Coordinator) Deliver(ctx context.Context, ev Event) error {
p := Partition(ev.OrderID, c.partitions)
c.mu.Lock()
if c.pending[p] == nil {
c.pending[p] = map[int64]*item{}
}
if ev.Seq <= c.expected[p] {
c.mu.Unlock()
return nil
}
if _, exists := c.pending[p][ev.Seq]; !exists {
c.pending[p][ev.Seq] = &item{ev: ev}
}
c.mu.Unlock()
for {
c.mu.Lock()
next := c.expected[p] + 1
it, ok := c.pending[p][next]
c.mu.Unlock()
if !ok {
return nil
}
err := c.handle(ctx, it.ev)
c.mu.Lock()
if err == nil {
delete(c.pending[p], next)
c.expected[p] = next
c.mu.Unlock()
continue
}
it.attempts++
if it.attempts >= c.maxAttempts {
c.dlq = append(c.dlq, DeadLetter{Event: it.ev, Attempts: it.attempts, Err: err.Error()})
delete(c.pending[p], next)
c.expected[p] = next
c.mu.Unlock()
return fmt.Errorf("event %s -> DLQ: %w", it.ev.ID, err)
}
c.mu.Unlock()
return fmt.Errorf("event %s attempt %d/%d: %w", it.ev.ID, it.attempts, c.maxAttempts, err)
}
}
func (c *Coordinator) BacklogAge(now time.Time) time.Duration {
c.mu.Lock()
defer c.mu.Unlock()
var oldest time.Time
for _, part := range c.pending {
for _, it := range part {
if oldest.IsZero() || it.ev.EnqueuedAt.Before(oldest) {
oldest = it.ev.EnqueuedAt
}
}
}
if oldest.IsZero() {
return 0
}
return now.Sub(oldest)
}
const (
BacklogWarnAge = 30 * time.Second
BacklogPageAge = 2 * time.Minute
)
func ClassifyBacklog(age time.Duration) string {
if age >= BacklogPageAge {
return "PAGE"
}
if age >= BacklogWarnAge {
return "WARN"
}
return "OK"
}
func (c *Coordinator) ReplayDLQ(ctx context.Context) (replayed, remaining int) {
c.mu.Lock()
batch := append([]DeadLetter(nil), c.dlq...)
c.dlq = nil
c.mu.Unlock()
var failed []DeadLetter
for _, d := range batch {
if err := c.handle(ctx, d.Event); err != nil {
d.Attempts++
d.Err = err.Error()
failed = append(failed, d)
continue
}
replayed++
}
c.mu.Lock()
c.dlq = append(c.dlq, failed...)
c.mu.Unlock()
return replayed, len(failed)
}
// file: internal/pipeline/reconcile.go
// tier: L1
// verify: go test ./internal/pipeline/
package pipeline
import (
"context"
"fmt"
"time"
)
type OrderSnapshot struct {
OrderID string
Status string
UpdatedAt time.Time
}
type ReconcileReport struct {
Scanned, SkippedFresh, Drifted, Emitted int
}
func expectedEvent(status string) string {
switch status {
case "INVENTORY_RESERVED":
return "inventory.reserved"
case "PAID":
return "payment.captured"
case "FULFILLED":
return "order.fulfilled"
default:
return ""
}
}
func Reconcile(ctx context.Context, now time.Time, window, grace time.Duration, orders []OrderSnapshot,
hasEvent func(ctx context.Context, orderID, kind string) (bool, error),
emit func(ctx context.Context, orderID, reason string) error) (ReconcileReport, error) {
var rep ReconcileReport
since := now.Add(-window)
for _, o := range orders {
if o.UpdatedAt.Before(since) {
continue
}
rep.Scanned++
// 刚更新的订单可能事件还在路上，先放过，避免对账和正常投递打架。
if grace > 0 && now.Sub(o.UpdatedAt) < grace {
rep.SkippedFresh++
continue
}
kind := expectedEvent(o.Status)
if kind == "" {
continue
}
ok, err := hasEvent(ctx, o.OrderID, kind)
if err != nil {
return rep, err
}
if ok {
continue
}
rep.Drifted++
if err := emit(ctx, o.OrderID, fmt.Sprintf("status=%s missing=%s", o.Status, kind)); err != nil {
return rep, err
}
rep.Emitted++
}
return rep, nil
}
测试：乱序、毒消息、重放与对账
一个测试串起四类场景：seq=2 先到必须等待；重复 seq=1 不重复执行；毒消息三次失败进 DLQ，修复后重放成功；BacklogAge 从 45 秒回到 0；对账只补偿窗口内、超过 grace、事件缺失的订单。
// file: internal/pipeline/pipeline_test.go
// tier: L1
// verify: go test -v ./internal/pipeline/
package pipeline
import (
"context"
"errors"
"fmt"
"testing"
"time"
)
var base = time.Date(2026, 10, 17, 16, 0, 0, 0, time.UTC)
func ev(orderID string, seq int64, at time.Time) Event {
return Event{ID: fmt.Sprintf("%s-%d", orderID, seq), OrderID: orderID, Seq: seq, EnqueuedAt: at}
}
func TestPipeline(t *testing.T) {
ctx := context.Background()
must := func(err error) {
t.Helper()
if err != nil {
t.Fatal(err)
}
}
// 乱序缓冲、按 seq 执行、重复不执行。
var got []int64
c := New(4, 3, func(_ context.Context, e Event) error { got = append(got, e.Seq); return nil })
must(c.Deliver(ctx, ev("o-1", 2, base)))
if len(got) != 0 {
t.Fatalf("seq 2 must wait, got %v", got)
}
must(c.Deliver(ctx, ev("o-1", 1, base)))
if len(got) != 2 || got[0] != 1 || got[1] != 2 {
t.Fatalf("order=%v want [1 2]", got)
}
must(c.Deliver(ctx, ev("o-1", 1, base)))
if len(got) != 2 {
t.Fatalf("duplicate applied: %v", got)
}
// 毒消息三次失败进 DLQ，修复后重放成功。
fail := true
var processed []int64
d := New(1, 3, func(_ context.Context, e Event) error {
if e.Seq == 1 && fail {
return errors.New("payload cannot be decoded")
}
processed = append(processed, e.Seq)
return nil
})
for i := 1; i <= 3; i++ {
if err := d.Deliver(ctx, ev("o-2", 1, base)); err == nil {
t.Fatalf("attempt %d: want error", i)
}
}
if len(d.dlq) != 1 {
t.Fatalf("dlq=%d want 1", len(d.dlq))
}
must(d.Deliver(ctx, ev("o-2", 2, base)))
if len(processed) != 1 || processed[0] != 2 {
t.Fatalf("processed=%v want [2]", processed)
}
if replayed, remaining := d.ReplayDLQ(ctx); replayed != 0 || remaining != 1 {
t.Fatalf("before fix: replayed=%d remaining=%d", replayed, remaining)
}
fail = false
if replayed, remaining := d.ReplayDLQ(ctx); replayed != 1 || remaining != 0 {
t.Fatalf("after fix: replayed=%d remaining=%d", replayed, remaining)
}
if len(processed) != 2 || processed[1] != 1 {
t.Fatalf("processed=%v want [2 1]", processed)
}
// 积压年龄与 30s/2min 两档阈值。
b := New(1, 3, func(context.Context, Event) error { return nil })
must(b.Deliver(ctx, ev("o-3", 2, base.Add(-45*time.Second))))
if age := b.BacklogAge(base); age != 45*time.Second || ClassifyBacklog(age) != "WARN" {
t.Fatalf("age=%s level=%s", age, ClassifyBacklog(age))
}
must(b.Deliver(ctx, ev("o-3", 1, base.Add(-46*time.Second))))
if age := b.BacklogAge(base); age != 0 {
t.Fatalf("drained age=%s want 0", age)
}
// 对账：窗口内、超过 grace、有预期事件但事件缺失的订单才补偿。
now := base
orders := []OrderSnapshot{
{OrderID: "paid-drift", Status: "PAID", UpdatedAt: now.Add(-10 * time.Minute)},
{OrderID: "paid-ok", Status: "PAID", UpdatedAt: now.Add(-10 * time.Minute)},
{OrderID: "created", Status: "CREATED", UpdatedAt: now.Add(-10 * time.Minute)},
{OrderID: "paid-fresh", Status: "PAID", UpdatedAt: now.Add(-30 * time.Second)},
{OrderID: "paid-old", Status: "PAID", UpdatedAt: now.Add(-30 * time.Hour)},
}
seen := map[string]bool{"paid-ok/payment.captured": true}
var emitted []string
hasEvent := func(_ context.Context, orderID, kind string) (bool, error) { return seen[orderID+"/"+kind], nil }
emit := func(_ context.Context, orderID, reason string) error {
emitted = append(emitted, orderID+"|"+reason)
return nil
}
rep, err := Reconcile(ctx, now, 24*time.Hour, 2*time.Minute, orders, hasEvent, emit)
must(err)
if rep.Scanned != 4 || rep.SkippedFresh != 1 || rep.Drifted != 1 || rep.Emitted != 1 {
t.Fatalf("report=%+v", rep)
}
if len(emitted) != 1 || emitted[0] != "paid-drift|status=PAID missing=payment.captured" {
t.Fatalf("emitted=%v", emitted)
}
}
真实运行结果
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），命令 go test -v ./... -count=1，原始输出保存在 .local/reports/be-arch/be-arch-perf-04-3-backlog-order-dlq-reconcile-tests.txt：
=== RUN TestPipeline
--- PASS: TestPipeline (0.00s)
PASS
ok archlab/internal/pipeline 0.487s
arch-lab 的阈值与对账口径
积压年龄 30 秒预警、2 分钟呼叫；积压条数只作为容量参考，不单独告警。DLQ 深度大于 0 即产生工单，重放前必须先确认处理器修复，否则只会把同一批消息再失败一次。对账每 5 分钟跑一次，窗口 24 小时，grace 2 分钟：窗口保证成本可控，grace 避开正常投递，超过 grace 仍缺事件才认定为漂移。补偿事件按 order_id + kind 幂等，重复对账不会产生重复副作用。
常见错误与边界
第一，用消息 ID 或随机数分区。同一订单落到不同分区，顺序保证失效；分区键必须是业务聚合 ID。
第二，只统计积压条数。大量短时间积压和少量长期积压的处理优先级完全不同，必须看最老事件年龄。
第三，毒消息无限重试。没有上限的重试会堵住整个分区，后续正常事件全部排队；到上限必须进 DLQ 并推进位点。
第四，对账直接改订单状态。对账是补事件，不是第二个写 owner；直接改状态会绕过状态机，制造更难查的冲突。
小结
积压、顺序、DLQ 和对账是异步链路的四个运维面：积压看年龄，顺序靠 order_id 分区加连续 seq，毒消息靠重试上限加 DLQ，最终一致性靠定时对账兜底。arch-lab 的实现覆盖了乱序、毒消息、重放和缺失事件四类测试。下一章转向资源隔离：连接池、线程池与协程池如何在过载时保护系统。
阅读导航：上一节：4.2 消息投递语义与消费幂等 · 下一节：5.1 连接池、线程池与协程池 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。