消息队列可靠性：把每一个故障窗口处理清楚

“消息已经发送成功，为什么订单还没有处理？”这个问题可能发生在生产者、Broker、消费者，也可能发生在消费者调用的数据库里。可靠性需要沿整条链检查，不能靠一个 ack=true 或“开启持久化”概括。

本文要解决六件事：消息怎样尽量不丢，重复怎样不变成重复业务，顺序在哪里成立，积压怎样恢复，坏消息怎样隔离，延迟任务怎样安全执行。以 Kafka 4.2、RabbitMQ 4.2 和 RocketMQ 5.x 文档解释具体边界。文中 Go 实验在 Go 1.27.2、Linux/amd64 运行，都是显式注入故障的内存模型；没有做真实 Broker 宕机或数据库提交实验。

从“消息不丢”改成一个能验证的承诺

先问清起点和终点。例如：“订单数据库成功提交后，这个订单最终会进入通知处理，允许重复投递，但同一通知只产生一个业务记录。”这个目标比“系统不丢消息”具体：订单提交是起点，通知业务落库是终点，中间的每次交接都要有可恢复证据。

发生故障的位置	可能留下的状态	必须有的恢复办法
订单提交后，发布前	数据库有订单，Broker 没消息	同事务 Outbox，或等价的可恢复发布记录
Broker 接收后，确认返回前	Broker 有消息，生产者不知道	用同一事件 ID 重试，接受可能重复
Broker 确认后，存储或副本故障	是否可恢复取决于确认与复制策略	明确持久化、副本数、故障域和故障模型
消费者取到后，业务提交前	消息在途，业务未完成	未确认消息可重新获取或重投
业务提交后，消费确认前	业务已完成，消息可能重来	原子地记录幂等结果，然后安全重复处理
消费确认后，业务提交前	进度已前移，业务未完成	改确认时点；已有缺口只能依赖补偿与对账

“超时”尤其容易被误读。它通常只说明调用方没及时得到确定结果，不能推出服务端没做。于是可靠系统需要一种“结果未知”的分支：保留事件、查状态或幂等重试，而不是把所有超时都当作发布失败。RabbitMQ 官方可靠性指南也要求生产者对未确认消息准备重发，并要求消费者能处理重复。^[1]

生产端确认、持久化和副本要一起看

生产端发送 API 返回 nil，可能只表示数据进了客户端缓冲区，也可能已经等到服务端确认。异步客户端通常还要检查逐条回调或 future。应用退出时需要按库的约定等待在途发布结束；直接退出可能丢掉还未发出的内存缓冲。

Kafka 普通生产者的 acks=all、Topic 的 min.insync.replicas 和实际 ISR 状态共同决定确认条件；副本因子为 3 不等于每条记录都等了 3 台，也不等于允许任意 2 台永久损坏。RabbitMQ 的 durable queue、persistent message 和 publisher confirm 分别解决不同环节；需要复制时还必须选对队列类型，例如 quorum queue。持久化和复制减少特定故障带来的丢失风险，不能覆盖管理员删除、磁盘集体损坏、保留期到期等所有情形。^[2][3][4]

RabbitMQ 还有路由这一层：一条消息可能被 exchange 接受，却没有路由到任何 queue。开启 confirm 后，无路由消息也可能得到确认。若业务要求至少进一个队列，发布者还应处理 mandatory 对应的返回消息，而非只盯 confirm。^[5]

先做业务，再确认，仍然会重复

设消费者处理一条 amount=100 的积分事件，先加积分，再 ack。业务完成后进程退出，ack 未到 Broker，下次会再加一次。

func unsafeReplay() int {
    balance := 0
    balance += 100 // 业务提交；确认之前退出
    balance += 100 // 重投后再次执行
    return balance
}
fmt.Println("unsafe balance:", unsafeReplay())

实际输出：

unsafe balance: 200

把 ack 放到前面并不能两全：它只是把风险换成“确认已成功，业务没来得及做”。投递语义需要说清对象：

语义	在指定边界内承诺什么	应用要付出的代价
At-most-once，至多一次	不依靠重投补齐，可能漏处理	接受损失；用于本来就允许丢的信号等
At-least-once，至少一次	在重试、保留和可用性等前提下持续交付，可能重复	业务幂等、去重、重试治理
Exactly-once，恰好一次	在明确参与协议的范围内，结果只提交一次	事务或去重协议及其约束；外部系统需另行协作

这里“至少一次”不意味着无限重试之后一定成功。永久坏消息、保留期到期、被删除的数据或长期停机都可能让它无法完成。Kafka 的事务能覆盖参与事务的 Kafka 记录和消费进度；它不会把第三方支付或邮件发送自动纳入同一个提交点。^[6]

幂等要挡住重复的业务效果

最实用的起点是稳定的事件 ID，以及消费端的 Inbox/处理记录。对同一消费者，以 (consumer_name, event_id) 建唯一约束，把“插入处理记录”和“更新业务状态”放在同一个数据库事务内。插入成功才执行这次业务；重复 ID 则返回已有结果。PostgreSQL 的唯一约束用于阻止并发重复键，事务用于把这两个数据库变化一起提交或一起撤销。^[7][8]

正在绘制图示…

可切换适合窗口或原始尺寸

查看图源代码

flowchart TD
    M["收到事件 e1"] --> T["开始本地数据库事务"]
    T --> I{"Inbox 唯一键是否已存在？"}
    I -- 否 --> B["写 Inbox + 更新业务状态"]
    B --> C["提交事务"]
    I -- 是 --> V["核对事件内容／读取已有结果"]
    V -- 内容一致 --> E["结束该分支的本地事务"]
    E --> A["确认消息"]
    V -- 内容冲突 --> Q["结束事务；隔离并报警"]
    C --> A
    B -- 失败 --> R["回滚；保留重试机会"]

放大图示

原始尺寸下可横向或纵向滚动；按 Escape 关闭。

正在绘制图示…

查看图源代码

flowchart TD
    M["收到事件 e1"] --> T["开始本地数据库事务"]
    T --> I{"Inbox 唯一键是否已存在？"}
    I -- 否 --> B["写 Inbox + 更新业务状态"]
    B --> C["提交事务"]
    I -- 是 --> V["核对事件内容／读取已有结果"]
    V -- 内容一致 --> E["结束该分支的本地事务"]
    E --> A["确认消息"]
    V -- 内容冲突 --> Q["结束事务；隔离并报警"]
    C --> A
    B -- 失败 --> R["回滚；保留重试机会"]

图中的已有记录必须表示“业务已提交”，不能在业务之前独立写一个 seen=true。否则业务失败，重试时却被当作重复跳过。同样，先查再写也不够：两个 worker 可以同时查到“不存在”，然后各执行一次。原子约束才是并发边界。

下面的 Store.Apply 用一把互斥锁模拟原子提交：先检查旧 ID 和参数，失败则不改任何状态，成功才一起修改余额和记录。它不提供磁盘持久化，只验证该模型中的状态不变量。

s := new(Store)
inserted, err := s.Apply(Event{"e1", 100}, false)
fmt.Println("first:", inserted, err)
inserted, err = s.Apply(Event{"e1", 100}, false)
fmt.Println("redelivery:", inserted, err)
_, err = s.Apply(Event{"e1", 200}, false)
fmt.Println("conflict:", err)
_, err = s.Apply(Event{"e2", 50}, true) // 注入提交前失败
fmt.Println("failed transaction:", err)
fmt.Println("safe balance:", s.balance)

first: true <nil>
redelivery: false <nil>
conflict: event ID reused with different payload
failed transaction: transaction aborted
safe balance: 100

同一个 ID 却换了金额，不应该悄悄当作成功重复。现实中可保存规范化内容的摘要，遇到冲突报警。去重记录的保留时间也要覆盖可能重放的时间范围，否则半年后重放的旧事件会重新生效。

“把余额设置为 100”看起来天然幂等，但若旧事件晚到，它会把新余额覆盖掉。状态覆盖通常还需要业务版本比较，例如仅接受比当前版本新的状态；增量事件则要按事件 ID 去重。幂等和顺序是相关问题，不能互相替代。

外部副作用不能塞进本地事务就算完成

数据库事务内部调用邮件或支付 API，并不能使那个 API 随数据库回滚。外部请求成功、数据库提交失败，下次仍然可能重复调用。可行方向是把外部执行意图再次写入 Outbox，由执行器用稳定幂等键调用支持该能力的外部服务，并保存查询、补偿和对账所需的信息。如果外部服务既不能幂等执行，也不能查询已有结果，就必须承认存在不可自动消除的结果未知窗口。

顺序需要同时约束生产、存储和执行

“按订单顺序处理”应定义为同一订单的状态变化顺序。全局顺序要求所有订单排成一条队，通常会牺牲并发能力，而且业务未必需要。

环节	保持单键顺序的常见办法	会破坏什么
生产	同一键使用一致路由，控制并发写入与重试	两个生产者同时发布并不自动获得业务先后关系
Broker	将相关记录放在同一有序单元，例如 Kafka 分区	不同分区之间没有天然全局顺序
消费	同键串行，或按业务版本检查依赖	拉取顺序不等于并发 worker 的完成顺序
失败重试	阻塞该键后续事件，或明确设计补偿/乱序处理	把失败事件送到重试队列，后续事件可能先完成

RabbitMQ 的重新入队、多个活动消费者和优先级也会改变应用观察到的完成顺序；队列文档对这些限制有明确说明。^[9] Kafka 非幂等生产者在存在多条在途请求时重试，也可能造成顺序变化，必须同时检查幂等和 in-flight 配置。^[2:1]

遇到毒消息时，这里有个真实取舍：继续阻塞可以保住该键后续依赖，却拖住进度；跳过可以恢复吞吐，却需要下游接受缺失事件。不要偷偷用“丢进 DLQ”代替这个业务决定。

积压：先算净处理能力，再决定是否扩容

积压本身未必异常，夜间批处理和短时削峰本来就允许排队。危险的是最老消息的年龄接近业务期限或日志保留期，以及处理能力长期低于进入速度。

稳定速率下，可以用一个简单估算：清空时间约等于“积压量 ÷（处理速度 − 进入速度）”。这是容量模型，不是某个 Broker 的性能结论。

backlog, incoming, capacity := 6000, 100, 150
fmt.Println("drain seconds:", backlog/(capacity-incoming))

drain seconds: 120

每秒处理 150 条并不意味着 6000 条只需 40 秒，因为每秒仍有 100 条进入。若处理速度不大于进入速度，按这个模型永远清不空；现实中重试还会增加额外流量。

排查时先拆三层：Broker 中待取的消息、客户端已取但未完成的消息、数据库或外部服务正在等待的请求。只看队列变短可能误判，因为消息只是被搬进消费者内存。建议同时观察消费 lag/ready、unacked/在途数、最老消息年龄、成功吞吐、重试量和下游延迟。

扩消费者前先查下游容量、热点分区和锁竞争。Kafka 普通组成员已经多于可分配分区时，继续加成员不会增加分区级并行度。若数据库已经被打满，再加 worker 可能让重试更多。批量、限流、优先恢复重要业务、暂缓低价值流量，都比盲目增加 goroutine 更有针对性。日志保留策略会删除旧消息，即使某个消费组还没有跟上，因此必须预留恢复窗口。^[3:1]

重试与死信：让坏消息离开正常通道

错误至少要分成三类：暂时不可用，可以等一等再试；内容或业务规则永久不合法，重复执行没有意义；结果未知，先查结果或用幂等键重试。这个分类属于应用策略，不会由 Broker 自动理解。

下面设定最多 3 次尝试，永久错误直接隔离。attempt 从 1 开始，达到 3 表示第三次已经失败。

fmt.Println("temporary attempt 1:", route(false, 1))
fmt.Println("temporary attempt 3:", route(false, 3))
fmt.Println("invalid payload:", route(true, 1))

temporary attempt 1: retry-later
temporary attempt 3: quarantine
invalid payload: quarantine

真实重试还需要退避和抖动，例如随失败次数增加等待时间，再在一段范围内随机分散。这是防止大量消费者同时冲击刚恢复服务的办法；最大尝试数、最长重试时长和超时预算应一起设置。代码里的路由函数只表达决策，没有实现定时调度。

死信队列（DLQ）用来保留退出正常处理流程的消息，便于检查、修复和受控重放。至少保留原事件 ID、原目的地、失败类别、首次/末次失败时间、尝试次数和可定位的错误信息，同时避免把令牌、完整敏感负载随意写进日志。DLQ 需要负责人、告警和重放流程；消息进去后无人处理，只是换了一个地方积压。

RabbitMQ 的 DLX 是交换机，DLQ 是它路由到的队列。拒绝且不重新入队、过期、超长等事件可以触发 dead-lettering。默认 dead-letter 转发并不自动具备安全的端到端交接，quorum queue 的 at-least-once dead-lettering 需要显式策略和配套限制；目标不可用时还会消耗源队列资源，并可能重复转发。^[10][4:1]

自己实现“先发布重试消息，再 ack 原消息”也有重复窗口；反过来先 ack 再发布又有丢失窗口。要么使用产品提供且正确配置的原子机制，要么用可靠中转记录和消费幂等处理这个裂缝。Kafka 普通 consumer group 的 offset 本身不是一套自动重试队列，重试 Topic、隔离 Topic 和进度关系需要应用定义。

延迟消息：到点可以开始，不保证到点已经完成

“30 分钟未支付就取消订单”应该表达为“30 分钟后检查订单是否仍满足取消条件”。支付可能先发生，延迟消息也可能重复或晚到，所以执行器要重新读状态，并用条件更新避免把已支付订单取消。

pending := []Scheduled{{"cancel-unpaid", 5}}
ready, pending := release(4, pending)
fmt.Println("t=4 ready:", ready)
ready, _ = release(8, pending)
fmt.Println("t=8 ready:", ready)

t=4 ready: []
t=8 ready: [cancel-unpaid]

这个教学调度器只在被调用时检查时间。任务在 t=5 到期，直到 t=8 才被检查，照样晚了 3 个时间单位。它展示“尚未到期不能释放”与“何时实际执行”是两个约束，不是一个。

产品边界也不相同。RocketMQ 5.x 延迟消息按投递时间戳安排可见性，允许范围与过期时间处理必须按当前服务版本校验。RabbitMQ 的 TTL 表示过期，结合 DLX 可搭建一些延迟转发方案，但 TTL 不是精确定时器，逐消息 TTL 在队列中的移除时机也有边界。不要把 Kafka 消息中的 timestamp 字段当成“在此时间之前不投递”的调度指令。^[11][12]

这类任务还要定义时钟来源、最长迟到容忍、取消与修改、恢复后补跑，以及什么时候仍应执行。业务已经过期的提醒，晚发可能比不发更糟。

可靠性最终要落成一份故障清单：哪个状态可重建，哪个提交点不能跨越，重复由哪里吸收，无法自动恢复时由谁处理。把这份清单写进测试和监控，比给系统贴“高可靠”的标签更有效。

依据与本次验证

资料于 2026-10-13 核对。下列官方文档支撑产品和数据库行为；重试分类、监控组合与容量估算是本文在这些边界上提出的工程建议。

Go 1.27.2、Linux/amd64 上，正常示例、100 个 goroutine 并发重复事件测试、提交前失败后重试测试全部通过；race 与 vet 通过。一个带 negative 构建标签的故意失败测试反驳“业务后 ack 就不会重复”的错误断言，实际输出包含：

--- FAIL: TestFalseClaimAckAfterBusinessPreventsDuplicates (0.00s)
    negative_test.go:9: counterexample: wanted one effect (100), got 200

该命令实际退出码为 1。测试只覆盖内存模型设定的故障窗口，没有验证数据库隔离级别、Broker 网络故障、真实重试时间和集群故障恢复。

完整可复现代码

所有文件放在同一目录。正常验证和故意失败反例分开执行：

go run .
go test -v ./...
go test -race ./...
go vet ./...
# 下面这一条预期失败，退出码为 1：
go test -tags negative -run TestFalseClaim ./...

go.mod

module example.com/mq/reliability

go 1.27.2

main.go

代码块 · 97 行 · 点击收起 / 展开

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Event struct {
	ID     string
	Amount int
}
type Store struct {
	mu      sync.Mutex
	balance int
	seen    map[string]int
}

// Apply models one atomic transaction using a mutex. It has no durability.
func (s *Store) Apply(e Event, failBeforeCommit bool) (bool, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if old, ok := s.seen[e.ID]; ok {
		if old != e.Amount {
			return false, errors.New("event ID reused with different payload")
		}
		return false, nil
	}
	if e.ID == "" || e.Amount <= 0 {
		return false, errors.New("invalid event")
	}
	if failBeforeCommit {
		return false, errors.New("transaction aborted")
	}
	if s.seen == nil {
		s.seen = make(map[string]int)
	}
	s.balance += e.Amount
	s.seen[e.ID] = e.Amount
	return true, nil
}

func unsafeReplay() int {
	balance := 0
	balance += 100 // business commit; crash before acknowledgement
	balance += 100 // redelivery repeats the business effect
	return balance
}

func route(permanent bool, attempt int) string {
	if permanent || attempt >= 3 {
		return "quarantine"
	}
	return "retry-later"
}

type Scheduled struct {
	ID  string
	Due int
}

func release(now int, pending []Scheduled) (ready []string, rest []Scheduled) {
	for _, m := range pending {
		if m.Due <= now {
			ready = append(ready, m.ID)
		} else {
			rest = append(rest, m)
		}
	}
	return
}

func demo() {
	fmt.Println("unsafe balance:", unsafeReplay())
	s := new(Store)
	inserted, err := s.Apply(Event{"e1", 100}, false)
	fmt.Println("first:", inserted, err)
	inserted, err = s.Apply(Event{"e1", 100}, false)
	fmt.Println("redelivery:", inserted, err)
	_, err = s.Apply(Event{"e1", 200}, false)
	fmt.Println("conflict:", err)
	_, err = s.Apply(Event{"e2", 50}, true)
	fmt.Println("failed transaction:", err)
	fmt.Println("safe balance:", s.balance)
	fmt.Println("temporary attempt 1:", route(false, 1))
	fmt.Println("temporary attempt 3:", route(false, 3))
	fmt.Println("invalid payload:", route(true, 1))
	pending := []Scheduled{{"cancel-unpaid", 5}}
	ready, pending := release(4, pending)
	fmt.Println("t=4 ready:", ready)
	ready, _ = release(8, pending)
	fmt.Println("t=8 ready:", ready)
	backlog, incoming, capacity := 6000, 100, 150
	fmt.Println("drain seconds:", backlog/(capacity-incoming))
}

func main() { demo() }

main_test.go

代码块 · 49 行 · 点击收起 / 展开

package main

import (
	"sync"
	"testing"
)

func Example() {
	demo()
	// Output:
	// unsafe balance: 200
	// first: true <nil>
	// redelivery: false <nil>
	// conflict: event ID reused with different payload
	// failed transaction: transaction aborted
	// safe balance: 100
	// temporary attempt 1: retry-later
	// temporary attempt 3: quarantine
	// invalid payload: quarantine
	// t=4 ready: []
	// t=8 ready: [cancel-unpaid]
	// drain seconds: 120
}

func TestConcurrentIdempotency(t *testing.T) {
	s := new(Store)
	var wg sync.WaitGroup
	for range 100 {
		wg.Go(func() {
			if _, err := s.Apply(Event{"e1", 100}, false); err != nil {
				t.Error(err)
			}
		})
	}
	wg.Wait()
	if s.balance != 100 {
		t.Fatalf("balance = %d", s.balance)
	}
}

func TestAbortCanRetry(t *testing.T) {
	s := new(Store)
	if _, err := s.Apply(Event{"e1", 100}, true); err == nil {
		t.Fatal("expected abort")
	}
	if applied, err := s.Apply(Event{"e1", 100}, false); !applied || err != nil {
		t.Fatalf("retry failed: %v %v", applied, err)
	}
}

negative_test.go

//go:build negative

package main

import "testing"

func TestFalseClaimAckAfterBusinessPreventsDuplicates(t *testing.T) {
	if got := unsafeReplay(); got != 100 {
		t.Fatalf("counterexample: wanted one effect (100), got %d", got)
	}
}

---

1. RabbitMQ 4.2 Reliability Guide。 ↩︎

2. Kafka 4.2 Producer Configs：acks、幂等、重试、在途请求与排序。 ↩︎ ↩︎

3. Kafka 4.2 Topic Configs：min.insync.replicas、保留与清理。 ↩︎ ↩︎

4. RabbitMQ 4.2 Quorum Queues：复制、毒消息与 at-least-once dead-lettering 的启用条件、代价和限制。 ↩︎ ↩︎

5. RabbitMQ 4.2 Confirms：生产确认、消费者确认、无法路由时的确认和 return。 ↩︎

6. Kafka 4.2 Design，Message Delivery Semantics / Using Transactions；KafkaConsumer 4.2，Manual Offset Control。 ↩︎

7. PostgreSQL 18 Constraints，Unique Constraints。 ↩︎

8. PostgreSQL 18 Transactions。 ↩︎

9. RabbitMQ 4.2 Queues，Message Ordering。 ↩︎

10. RabbitMQ 4.2 Dead Letter Exchanges：触发原因、重新发布和默认安全边界。 ↩︎

11. RocketMQ 5.0 Delay Message：时间戳、允许范围、投递与执行边界。本文不把文档中的默认范围当成所有托管发行版的承诺。 ↩︎

12. RabbitMQ 4.2 TTL：消息到期与移除规则。 ↩︎