主题
消息队列应用模式：从一次订单提交走到可靠的异步处理
创建订单和通知仓库，看起来只是先写数据库、再发一条消息。但进程可以恰好在两步之间退出。仓库处理到一半也可能退出，或者消息已发送、确认却没回来。事件驱动架构是否可靠，取决于这些中间状态有没有留下可恢复的证据。
本文沿着这条链解释四种常见做法：事件驱动、本地消息表/Outbox、CDC 和任务队列。它们不在同一层：事件驱动描述服务如何协作，Outbox 保护本地业务提交与待发布事件之间的关系，CDC 负责捕获已提交变化，任务队列管理要执行的工作。读完应能判断该把哪一段交给哪种机制，以及还剩哪些重复和失败窗口。
文档依据于 2026-10-13 核对，数据库说明采用 PostgreSQL 18，CDC 路由说明采用 Debezium 3.3 作为固定讲解基线。Go 实验在 Go 1.27.2、Linux/amd64 运行，全部是单线程、显式故障注入的内存模型；没有运行数据库、CDC 连接器或 Broker。
事件、命令和任务，先分清谁在表达什么
| 消息 | 表达的意思 | 谁对处理负责 | 例子 |
|---|---|---|---|
| 领域事件 | 一件已发生的业务事实 | 生产者负责事实正确；各订阅者决定自己的反应 | OrderPaid |
| 命令 | 请求某个能力做一件事 | 明确的接收方决定接受、拒绝或执行 | ReserveInventory |
| 后台任务 | 一次有参数、执行状态和重试策略的工作 | 任务执行器负责调度和完成状态 | GenerateInvoicePDF |
支付完成以后发布 OrderPaid，通知、报表、仓库可以各自订阅；生产者不必同步调用所有下游。若业务要求“只有仓库明确预留成功才向用户承诺”，那就需要观察命令结果或设计跨服务流程，不能只发一个事件便宣布整体成功。
事件驱动降低的是服务在调用时机和具体接收者上的依赖，同时增加了事件契约、状态追踪、重试、最终一致性和运维工作。用户界面也需要表达“订单已接收，发货准备中”，而不是把异步完成伪装成同步完成。本文使用的事件/任务划分是设计上的责任约定，并不要求某一种 Broker。
事件驱动也不等于事件溯源（Event Sourcing）。普通系统可以用当前状态表作为权威数据，同时发领域事件；事件溯源则以事件历史作为重建状态的重要依据，还要另外处理完整历史、重放和演进。仅仅把几条消息送进 Kafka，并不会自动完成这些设计。
双写的裂缝：调整先后顺序不能消除故障
先提交订单再发消息，可能留下有订单没消息；先发消息再提交订单，可能产生数据库中从未成立的订单事件。网络调用超时还会让应用不知道对方到底做没做。
正在绘制图示…
查看图源代码
flowchart LR
A["方案一：提交订单"] --> B["退出发生在这里"]
B -. "没有机会执行" .-> C["发布事件"]
D["方案二：发布事件"] --> E["订单事务回滚"]
E --> F["下游却可能已收到事件"]
这不是给发送调用多加两次重试就能完整解决的问题。没有持久“待发送”记录，进程重启后可能连要重试什么都不知道。要保护这个交接，需要把业务变化和发布意图放到同一个能原子提交的存储边界里。
Outbox：把“稍后要发什么”写进同一个本地事务
本地消息表通常就是 Outbox 的一种实现。创建订单时，在同一个数据库事务里插入订单和 Outbox 事件行。事务回滚，两者都不成立；提交成功，后台 relay 即使暂时停机，仍有记录可供之后发送。PostgreSQL 的事务原子性支持的是这些同库变化，不能顺便把任意 Broker 网络操作也纳入事务。[1][2]
| Outbox 字段 | 用途 | 常见误用 |
|---|---|---|
event_id | 重发时保持不变，供下游去重 | 每次重试重新生成 ID，变成新事件 |
aggregate_type / aggregate_id | 表明哪类业务实体，提供分区或路由键 | 用事件 UUID 分区，拆散同订单事件 |
event_type / schema_version | 让消费者选择解释方式 | 改字段含义却沿用旧版本 |
payload | 保存发布时的事件内容 | 发送时重新读当前订单，把历史事实改成最新状态 |
| 创建时间与业务序号 | 观测等待时长、辅助单实体排序 | 把多事务自增 ID 直接当成提交顺序 |
| relay 状态或租约字段 | 支持轮询领取、重试、恢复 | 对 CDC 方案生硬套上不需要的状态更新流 |
Debezium 的 Outbox Event Router 有一套默认字段映射，可以把事件 ID 放入 header、aggregate ID 作为 Kafka key。上表是应用设计视角，字段名不全等于它的默认 schema，采用连接器时应按其配置明确映射。[2:1]
正在绘制图示…
查看图源代码
flowchart TB
API["订单请求"] --> TX["同一数据库事务"]
TX --> O["orders：订单状态"]
TX --> E["outbox：待发布事件"]
E --> R["relay：轮询或 CDC"]
R --> B["消息系统"]
B --> C["消费者"]
C --> IT["另一数据库事务<br/>Inbox + 业务变化"]
事务只封住了第一道裂缝
模型的 Create 在失败时不改变订单和 Outbox，成功时一起改变。实际输出先验证这个不变量：
go
d := new(DB)
fmt.Println("create failed:", d.Create("o0", true))
fmt.Println("after abort orders/outbox:", len(d.orders), len(d.outbox))text
create failed: transaction aborted
after abort orders/outbox: 0 0然后创建一条正常订单，让 relay 在发布之后、标记已发送之前“退出”：
go
if err := d.Create("o1", false); err != nil { panic(err) }
b := new(Broker)
d.Relay(b, true)
fmt.Println("after publish crash deliveries/sent:", len(b.Delivered), d.outbox[0].Sent)
d.Relay(b, false) // 重启后再次扫描同一行
p := new(Projection)
for _, e := range b.Delivered { p.Apply(e) }
fmt.Println("after restart deliveries/effects:", len(b.Delivered), p.count)text
after publish crash deliveries/sent: 1 false
after restart deliveries/effects: 2 1第一次已经送入模型 Broker，但 Outbox 还显示未发送；恢复后只能安全地重发，因此实际投递两次。消费者按同一个事件 ID 去重，最终只产生一次模型中的计数效果。Outbox 的价值是留下可恢复的发布意图，不是自动把数据库和 Broker 变成一个恰好一次事务。
这里保留内存对象来模拟“数据库状态在重启后仍存在”，并没有真正杀进程或落盘。Projection.Apply 也只有单线程计数。真实消费者必须用持久唯一键和本地事务一起保护去重记录与业务变化；简单的内存 map 或“先写已处理、再改业务”不能抵抗真实崩溃。
轮询 relay 怎样领取工作
最直接的 relay 定期查未发送行，领取一小批，发布并等待对应确认，然后标记完成。多个 relay 并行时，要避免无约束地抢同一行，常见做法是行锁，或带到期时间的租约。
PostgreSQL 的 FOR UPDATE SKIP LOCKED 可让 worker 跳过其他事务已锁定的行，适合队列式领取；官方文档也指出它提供的是不一致视图，不能当通用查询的一致性方案。[3]
一种实现把锁持有到发布和状态更新结束，简单但网络慢时会拖长数据库事务；另一种先在短事务中写入租约，提交后再发，完成时按领取者和版本条件更新，减少锁时间但增加租约恢复逻辑。两者都存在“发布成功、完成标记失败”的重复窗口。
还要专门处理顺序。两个 relay 用 SKIP LOCKED 领取同一订单的不同事件时，后面的事件可能先发。自增 ID 的分配顺序也不必等于并发事务提交顺序。需要单实体顺序时，可以按 aggregate 分配单一执行通道，使用明确业务序号校验，并让下游处理重投。不要以为 SQL 加了 ORDER BY id 就覆盖了多 worker 的网络完成顺序。
轮询的优点是依赖较少、状态可直接查；缺点是扫描负载、索引与清理、轮询延迟，以及自己维护 relay。Outbox 清理必须等到事件可靠交接且审计/回放窗口已满足，不能仅按“创建满一天”删除尚未发布的行。
CDC：从提交日志捕获变化，不再反复扫业务表
CDC 是 Change Data Capture。基于日志的 CDC 从数据库的变更日志得到已提交变化，例如 PostgreSQL WAL 的逻辑解码。Debezium PostgreSQL connector 通常先做一致性快照，再从记录的位置持续读取后续变化；具体快照模式和恢复流程由配置决定。[4][5]
这里有两条不同路线：
| 捕获什么 | 发出的语义 | 适合哪里 | 需要承受的耦合 |
|---|---|---|---|
| 直接捕获业务表 | 某一行新增、更新、删除，可能含 before/after | 搜索索引、数据同步、分析入湖 | 下游依赖表结构和数据库变更含义 |
| 捕获专门 Outbox 表 | 应用主动写出的领域事件 | 跨服务业务协作 | 应用维护事件契约；CDC 负责可靠搬运 |
订单行的 status 从 2 变成 3，不一定足以告诉下游“付款通过哪种校验、是不是人工修正”。直接把所有行变化都叫业务事件，会迫使下游反向理解数据库内部实现。Outbox + CDC 则由应用在业务事务里明确写出 OrderPaid，连接器只读取和路由这个事件。
CDC 并没有移除重复问题。连接器输出成功、恢复位置还没持久化就故障，重启后可能再次输出。初始快照和流式变更的切换、重建连接器、手工重置 offset 都需要明确恢复语义。PostgreSQL 逻辑复制槽在崩溃恢复后也可能重新发送变化，客户端要处理重复。[5:1][4:1]
运维上尤其要观察源数据库日志保留。连接器长期停机或不能前移，复制槽可能让 WAL 无法正常回收，消耗磁盘；反过来，如果需要的日志已经不再可用，连接器可能无法从旧位置直接继续，需要重新快照或补数。这是数据库与消息平台共同的恢复问题，不能只监控 Kafka consumer lag。
一条可用的 CDC 链至少要监控：源端日志/槽进度、连接器状态和错误、快照进度、输出延迟、转换失败、目标积压，以及事件 ID 是否被消费者正确去重。权限也要最小化，别因为能抓整张表，就把密码哈希、令牌或不需要的个人数据发到多个下游。
事件契约：今天能读，还要考虑明年重放
一条可维护的事件通常需要稳定 ID、事件类型、来源、发生时间、业务实体 ID、schema 版本，以及解释事件所需的数据。CloudEvents 规范统一了 id、source、type、specversion 等信封属性；它的 specversion 表示 CloudEvents 规范版本，不应拿来充当业务 payload 的版本号。[6]
正文模型使用自定义小信封，并非完整 CloudEvents：
json
{"id":"e1","type":"order.created","version":1,"order_id":"o1"}兼容性要用消费者真实行为验证。Go 的 encoding/json 默认忽略结构体中不存在的 JSON 字段，适合有约定的可选字段增量扩展，但也可能吞掉拼写错误。下面的解码器允许额外字段，同时显式验证必填身份字段、事件类型和支持的版本：
go
_, err := Decode([]byte(`{"id":"e1","type":"order.created","version":1,"order_id":"o1","region":"eu"}`))
fmt.Println("additive field:", err)
_, err = Decode([]byte(`{"id":"e2","type":"order.created","version":2,"order_id":"o1"}`))
fmt.Println("new contract:", err)text
additive field: <nil>
new contract: unsupported event contract第一条能读只是这个消费者的实际行为，不能推导“新增字段永远兼容”。老消费者若禁止未知字段，或者新字段改变已有业务含义，就会破坏兼容。未知枚举、字段改类型、单位从分改成元、可空性改变，往往比加字段更危险。[7]
发布新契约时，先让消费者能接受过渡形态，再逐步切生产者；保留旧事件的解释代码或受控转换工具。重放时还要隔离外部副作用：用于重建报表的旧支付事件，不应该再次给用户发送“刚支付成功”的通知。事件日志能重读，不等于每种业务动作都应该再执行。
任务队列：把“谁领走、何时重试”变成显式状态
事件告诉订阅者发生了什么，任务队列负责一件工作有没有完成。典型任务会经历等待、运行、成功、等待重试、失败隔离或取消等状态，还要记录尝试次数、执行截止时间、下次运行时间和错误信息。
正在绘制图示…
查看图源代码
stateDiagram-v2
[*] --> Ready: 入队提交
Ready --> Running: worker 领取／获得租约
Running --> Done: 当前领取者确认完成
Running --> RetryWait: 暂时失败或租约到期
RetryWait --> Ready: 到达下次尝试时间
Running --> Quarantined: 永久失败或次数耗尽
Ready --> Cancelled: 取消待执行工作
Done --> [*]
Quarantined --> Ready: 修复后受控重试
取消待执行任务和中止正在执行的外部动作是两件事。worker 收到取消信号时，HTTP 请求可能已经被对方处理；租约到期后新 worker 领取，旧 worker 也可能只是暂停后恢复。任务系统需要幂等键、合理超时、心跳/续租，以及条件更新来拒绝过期领取者的完成报告。
下面模型每次领取增加一个 token，完成时同时检查 token 和租约时间：
go
l := new(Lease)
old, _ := l.Claim("worker-a", 0, 5)
fresh, _ := l.Claim("worker-b", 6, 5)
fmt.Println("stale completion:", l.Complete(old, 7))
fmt.Println("current completion:", l.Complete(fresh, 7))text
stale completion: false
current completion: trueA 的旧 token 不能再把任务标成成功。这个检查只保护模型的任务状态；如果 A 已经向外部银行发了请求，任务表拒绝它的报告也撤不回请求。真正的 fencing 需要被保护的资源也验证所有权版本，或者外部 API 使用稳定幂等键。模型没有模拟网络和外部资源，不应被当作完整分布式锁。
选择数据库任务队列时，还有一个实用优势：任务插入与业务更新可以复用同一个数据库事务。比如 River 的事务性 enqueue 接受已有事务，使“创建订单”和“排入后台任务”一起提交；但后续 worker 的执行仍可能重试，外部副作用仍要幂等。[8]
选择 Broker 任务队列时，要检查哪些能力由库提供，哪些由应用补：退避、并发限制、优先级、定时、失败保留、任务唯一性、监控和手工重放。不要把一个消费循环已经能运行，误认成任务生命周期已设计完整。
怎样选择组合，而不是叠满所有模式
如果只是把同库里的报表生成移到后台，事务性数据库任务队列可能已经够用。如果多个独立服务需要可靠接收领域事实，Outbox 加 relay 和消费端 Inbox 更贴合责任边界。如果已有 CDC 运维体系且要捕获持续变化，Outbox + CDC 能减少自己写轮询器的工作。需要跨服务等待、补偿和长期状态编排时，再评估工作流系统。
每多一层都需要回答一个具体问题。数据库负责哪次原子提交，relay 负责哪次交接，消息系统保留多久，消费者怎样识别重复，任务系统怎样恢复未完成工作。能把这些问题说清楚，才算把“异步”变成可维护的设计。
依据与本地验证
本机 go run .、go test -v ./...、go test -race ./...、go vet ./... 均通过。测试验证模型重发保留事件身份、完成行不继续发布，以及过期租约不能提交完成。没有数据库、relay 进程重启、CDC 或 Broker 集成测试；本文讨论的 SQL 领取与数据库事务实现属于官方行为基础上的设计说明，未伪装成已运行 SQL。
完整可复现代码
在同一目录保存下面文件，执行：
sh
go run .
go test -v ./...
go test -race ./...
go vet ./...go.mod
go.mod
module example.com/mq/patterns
go 1.27.2main.go
代码块 · 135 行 · 点击收起 / 展开
go
package main
import (
"encoding/json"
"errors"
"fmt"
)
type Event struct {
ID string `json:"id"`
Type string `json:"type"`
Version int `json:"version"`
OrderID string `json:"order_id"`
}
type Row struct {
Event Event
Sent bool
}
type DB struct {
orders map[string]bool
outbox []Row
}
// Create is a single-threaded atomic-state model, not a durable database.
func (d *DB) Create(order string, abort bool) error {
if order == "" || d.orders[order] {
return errors.New("invalid or duplicate order")
}
e := Event{ID: "created:" + order, Type: "order.created", Version: 1, OrderID: order}
if abort {
return errors.New("transaction aborted")
}
if d.orders == nil {
d.orders = make(map[string]bool)
}
d.orders[order] = true
d.outbox = append(d.outbox, Row{Event: e})
return nil
}
type Broker struct{ Delivered []Event }
func (d *DB) Relay(b *Broker, crashAfterPublish bool) {
for i := range d.outbox {
if d.outbox[i].Sent {
continue
}
b.Delivered = append(b.Delivered, d.outbox[i].Event)
if crashAfterPublish {
return
} // restart retains DB state but loses execution point
d.outbox[i].Sent = true
}
}
type Projection struct {
seen map[string]bool
count int
}
func (p *Projection) Apply(e Event) {
if p.seen[e.ID] {
return
}
if p.seen == nil {
p.seen = make(map[string]bool)
}
p.seen[e.ID] = true
p.count++
}
func Decode(b []byte) (Event, error) {
var e Event
if err := json.Unmarshal(b, &e); err != nil {
return e, err
}
if e.Type != "order.created" || e.Version != 1 {
return e, errors.New("unsupported event contract")
}
if e.ID == "" || e.OrderID == "" {
return e, errors.New("missing event identity")
}
return e, nil
}
type Lease struct {
Token int
Owner string
Until int
Done bool
}
func (l *Lease) Claim(owner string, now, duration int) (int, bool) {
if l.Done || now < l.Until {
return 0, false
}
l.Token++
l.Owner, l.Until = owner, now+duration
return l.Token, true
}
func (l *Lease) Complete(token, now int) bool {
if l.Done || token != l.Token || now >= l.Until {
return false
}
l.Done = true
return true
}
func demo() {
d := new(DB)
fmt.Println("create failed:", d.Create("o0", true))
fmt.Println("after abort orders/outbox:", len(d.orders), len(d.outbox))
if err := d.Create("o1", false); err != nil {
panic(err)
}
b := new(Broker)
d.Relay(b, true)
fmt.Println("after publish crash deliveries/sent:", len(b.Delivered), d.outbox[0].Sent)
d.Relay(b, false)
p := new(Projection)
for _, e := range b.Delivered {
p.Apply(e)
}
fmt.Println("after restart deliveries/effects:", len(b.Delivered), p.count)
_, err := Decode([]byte(`{"id":"e1","type":"order.created","version":1,"order_id":"o1","region":"eu"}`))
fmt.Println("additive field:", err)
_, err = Decode([]byte(`{"id":"e2","type":"order.created","version":2,"order_id":"o1"}`))
fmt.Println("new contract:", err)
l := new(Lease)
old, _ := l.Claim("worker-a", 0, 5)
fresh, _ := l.Claim("worker-b", 6, 5)
fmt.Println("stale completion:", l.Complete(old, 7))
fmt.Println("current completion:", l.Complete(fresh, 7))
}
func main() { demo() }main_test.go
代码块 · 38 行 · 点击收起 / 展开
go
package main
import "testing"
func Example() {
demo()
// Output:
// create failed: transaction aborted
// after abort orders/outbox: 0 0
// after publish crash deliveries/sent: 1 false
// after restart deliveries/effects: 2 1
// additive field: <nil>
// new contract: unsupported event contract
// stale completion: false
// current completion: true
}
func TestOutboxRetryKeepsIdentity(t *testing.T) {
d, b := new(DB), new(Broker)
if err := d.Create("o1", false); err != nil {
t.Fatal(err)
}
d.Relay(b, true)
d.Relay(b, false)
if len(b.Delivered) != 2 || b.Delivered[0] != b.Delivered[1] {
t.Fatal(b.Delivered)
}
d.Relay(b, false)
if len(b.Delivered) != 2 {
t.Fatal("sent row was published again")
}
}
func TestExpiredLeaseCannotComplete(t *testing.T) {
l := new(Lease)
token, _ := l.Claim("a", 0, 5)
if l.Complete(token, 5) {
t.Fatal("expired lease completed")
}
}PostgreSQL 18 Transactions：本地事务的原子性。 ↩︎
Debezium 3.3 Outbox Event Router：事件表字段、路由和去重 ID。采用固定讲解基线，不代表最新连接器版本。 ↩︎ ↩︎
PostgreSQL 18 SELECT，Locking Clause：
SKIP LOCKED的队列用途和不一致视图限制。 ↩︎PostgreSQL 18 Logical Decoding Concepts：逻辑复制槽、日志保留与崩溃后的重复输出。 ↩︎ ↩︎
Debezium 3.3 PostgreSQL connector：快照、流式变更、offset、WAL 与恢复条件。 ↩︎ ↩︎
CloudEvents 1.0.2 规范：信封属性和 schema 含义。 ↩︎
Go encoding/json：结构体解码、未知字段及解码约定；本文实际使用 Go 1.27.2 的 v1 API。 ↩︎
River Transactional enqueueing：复用数据库事务的入队。 ↩︎