本地事务能保证什么
6.1 确定了 orders 与 order_items 的写 owner 是订单服务。接下来要回答：状态迁移成功之后，如何可靠地通知库存、支付和读模型。直觉做法是“先更新数据库，再发一条 Kafka 消息”，但这两个动作不在同一个事务里：数据库提交成功、发送失败，下游永远不知道订单已经支付；发送成功、数据库回滚，下游收到一条不存在的状态。
站内 分布式数据一致性 已经系统讲过 2PC、TCC、Saga 的取舍，本节不重复协议对比，只处理 arch-lab 的最小闭环：一个数据库、一个订单服务、一个消息代理。结论先说：把“状态变更”和“待发消息”写进同一个本地事务，再用一个独立的投递器把消息送出去。这就是 outbox。
本地事务只保证一件事：同一数据库内，多条写入要么全部可见，要么全部不可见。它不保证消息已经到达代理，也不保证消费者只处理一次。后面两点必须由投递器和消费端分别承担。
另一个容易混淆的点是隔离级别。PostgreSQL 默认的 Read Committed 足以支撑“按版本条件更新”：UPDATE orders SET status = $1, version = version + 1 WHERE id = $2 AND version = $3 影响行数为 0 就返回冲突。不要为了 outbox 把整个库改成 Serializable；那会把并发写变成重试风暴，而版本条件已经表达了业务需要的串行化。
为什么不能先写库再发消息
“先写库、后发消息”的问题不是偶发失败，而是两个系统之间没有共同的提交点。订单服务在数据库提交后崩溃，消息没发；或者消息发出后数据库提交失败，消息已经在代理里。前者需要扫描补偿，后者需要下游回查，两个方向都会把一致性责任推给最不了解现场的一方。
在本地事务里直接调用 Kafka 客户端也不行。网络调用可能成功但响应丢失，事务无法感知；把网络 I/O 放进数据库事务还会拉长持锁时间，放大死锁概率。正确边界是：事务内只写本地表，事务外只做投递。
outbox 表结构
outbox 是订单库里的普通表，和 orders 同库同事务。目标结构如下（本机未启动 PostgreSQL，DDL 的真实执行属于 L3）：
| 列 | 类型 | 约束与用途 |
|---|---|---|
id | text | 主键，事件 ID，消费端去重键 |
aggregate_id | text | 聚合 ID，这里是订单号 |
aggregate_version | bigint | 聚合版本，配合唯一约束防重复事件 |
topic | text | 消息主题，如 order.paid |
payload | jsonb | 事件体，保留原始事实 |
created_at | timestamptz | 入队时间，投递按它排序 |
sent_at | timestamptz | 为空表示待投递 |
attempts | integer | 失败次数，用于退避与告警 |
唯一约束是 (aggregate_id, aggregate_version, topic)：同一聚合的同一版本只能产生一条同主题消息，重试写入会直接冲突，而不是悄悄多出一条。待投递扫描用部分索引 WHERE sent_at IS NULL，避免已发送的历史消息拖慢队列。
状态与消息同事务写入
RecordPayment 只接收一个 UnitOfWork，事务内先改状态、再追加消息。任何一步返回错误，整个事务回滚：
// file: internal/outbox/outbox.go
// tier: L1
// verify: go test ./internal/outbox/
package outbox
import (
"context"
"errors"
"fmt"
"sync"
"time"
)
var ErrAlreadyPending = errors.New("outbox message already pending")
var ErrNoOrder = errors.New("order not found")
type Message struct {
ID string
AggregateID string
AggregateVersion int
Topic string
Payload []byte
CreatedAt time.Time
SentAt *time.Time
Attempts int
}
type Writer interface {
SetOrderStatus(ctx context.Context, orderID, status string, version int) error
Append(ctx context.Context, m Message) error
}
type UnitOfWork interface {
InTx(ctx context.Context, fn func(Writer) error) error
}
type Store interface {
Pending(ctx context.Context, limit int) ([]Message, error)
MarkSent(ctx context.Context, id string) error
MarkFailed(ctx context.Context, id string, cause error) error
}
type Publisher interface {
Publish(ctx context.Context, m Message) error
}
// RecordPayment 在同一个事务里改状态并写 outbox。
// 任意一步失败，回滚后调用方看不到半成品状态。
func RecordPayment(ctx context.Context, uow UnitOfWork, orderID, providerRef string, now time.Time) error {
payload := []byte(fmt.Sprintf(`{"order_id":%q,"status":"PAID","provider_ref":%q}`, orderID, providerRef))
return uow.InTx(ctx, func(tx Writer) error {
if err := tx.SetOrderStatus(ctx, orderID, "PAID", 2); err != nil {
return err
}
return tx.Append(ctx, Message{
ID: "evt-" + orderID + "-paid", AggregateID: orderID,
AggregateVersion: 2, Topic: "order.paid", Payload: payload, CreatedAt: now,
})
})
}
type Dispatcher struct {
Store Store
Publisher Publisher
}
// RunOnce 只扫一轮。Publish 成功但 MarkSent 失败或进程崩溃时，
// 消息仍是 pending，下一轮会重复投递；这是有意的至少一次语义。
func (d *Dispatcher) RunOnce(ctx context.Context, limit int) (int, error) {
msgs, err := d.Store.Pending(ctx, limit)
if err != nil {
return 0, err
}
sent := 0
for _, m := range msgs {
if err := d.Publisher.Publish(ctx, m); err != nil {
if ferr := d.Store.MarkFailed(ctx, m.ID, err); ferr != nil {
return sent, ferr
}
continue
}
if err := d.Store.MarkSent(ctx, m.ID); err != nil {
return sent, err
}
sent++
}
return sent, nil
}
type Handler func(context.Context, Message) error
// Deduper 是 L1 去重：以 message_id 为键记录已成功处理的消费。
// 生产实现应把 (consumer, message_id) 放进唯一索引。
type Deduper struct {
mu sync.Mutex
seen map[string]struct{}
handle Handler
}
func NewDeduper(h Handler) *Deduper { return &Deduper{seen: map[string]struct{}{}, handle: h} }
func (d *Deduper) Handle(ctx context.Context, m Message) error {
d.mu.Lock()
defer d.mu.Unlock()
if _, ok := d.seen[m.ID]; ok {
return nil
}
if err := d.handle(ctx, m); err != nil {
return err
}
d.seen[m.ID] = struct{}{}
return nil
}
SetOrderStatus 和 Append 必须共享同一个 sql.Tx。在 PostgreSQL 适配器里，它们就是同一个事务对象上的两条 Exec；在内存实现里，用快照复制模拟同样的可见性。事务提交之前，外部读不到新状态，也读不到新消息。
投递器：至少一次
RunOnce 的语义是“至少一次”，不是“恰好一次”。消息先 Publish，成功后 MarkSent；如果进程在两者之间崩溃，sent_at 仍为空，下一轮会再次投递。代理侧看到的重复，必须由消费端消化。
这不是实现缺陷，而是分布式系统的现实：跨两个系统的“发送并确认”无法原子完成。要追求恰好一次，只能在消费端做幂等，或者让下游用业务键覆盖写。投递器能做的，是把重复窗口缩到最小，并把失败次数记进 attempts。
扫描频率也是取舍：太密会空转，太疏会放大延迟。L1 版本由调用方决定 RunOnce 的节奏；生产实现通常用固定间隔加指数退避，并把连续失败的 message 移到 dead-letter 表。dead-letter 不是删除，保留原 payload 与最后一次错误，才能人工重放。
消费端去重
Deduper 在调用业务处理器之前检查 message_id，成功后记录。同一事件重复投递三次，处理器只执行一次。L1 版本用进程内 map，只适合单实例演示；生产实现要把 (consumer, message_id) 写进唯一索引，并在同一事务里完成“业务写入 + 去重记录”。
顺序也要说清楚：先去重记录再执行业务，失败时删除记录，还是先执行业务再记录？前者要求业务失败可回滚，后者在崩溃时会重复执行。arch-lab 选择“业务写入与去重记录同事务”，因此去重表必须和业务表在同一个库；跨库时只能退化为业务幂等。
去重表还要有保留期。无限增长的记录会拖慢写入，通常按事件时间保留 7 到 30 天，超过窗口的重复投递只能靠业务幂等兜底；保留期必须大于消息代理的最长重投窗口。
测试：回滚、重复投递、重复消费
测试用内存快照模拟事务，覆盖三个关键断言：追加 outbox 失败时状态不前进；首次发布结果不明时消息留在 pending 并再次投递；同一 message_id 只触发一次处理器。
// file: internal/outbox/outbox_test.go
// tier: L1
// verify: go test ./internal/outbox/
package outbox
import (
"context"
"errors"
"maps"
"testing"
"time"
)
type memoryUoW struct {
states map[string]string
msgs map[string]Message
failAppend bool
}
func newMemoryUoW(orderID string) *memoryUoW {
return &memoryUoW{
states: map[string]string{orderID: "CREATED"},
msgs: make(map[string]Message),
}
}
// InTx 用快照模拟数据库事务：fn 失败时状态与出站消息一起丢弃。
func (u *memoryUoW) InTx(ctx context.Context, fn func(Writer) error) error {
states, msgs := u.states, u.msgs
u.states, u.msgs = maps.Clone(states), maps.Clone(msgs)
if err := fn(u); err != nil {
u.states, u.msgs = states, msgs
return err
}
return nil
}
func (u *memoryUoW) SetOrderStatus(ctx context.Context, orderID, status string, version int) error {
if _, ok := u.states[orderID]; !ok {
return ErrNoOrder
}
u.states[orderID] = status
return nil
}
func (u *memoryUoW) Append(ctx context.Context, m Message) error {
if u.failAppend {
return errors.New("simulated outbox insert failure")
}
if _, ok := u.msgs[m.ID]; ok {
return ErrAlreadyPending
}
u.msgs[m.ID] = m
return nil
}
func (u *memoryUoW) Pending(ctx context.Context, limit int) ([]Message, error) {
out := make([]Message, 0, len(u.msgs))
for _, m := range u.msgs {
if m.SentAt == nil {
out = append(out, m)
}
}
return out, nil
}
func (u *memoryUoW) MarkSent(ctx context.Context, id string) error {
m := u.msgs[id]
now := time.Now()
m.SentAt = &now
u.msgs[id] = m
return nil
}
func (u *memoryUoW) MarkFailed(ctx context.Context, id string, cause error) error {
m := u.msgs[id]
m.Attempts++
u.msgs[id] = m
return nil
}
func TestAtomicity(t *testing.T) {
ctx := context.Background()
uow := newMemoryUoW("o-1")
if err := RecordPayment(ctx, uow, "o-1", "pay-123", time.Now()); err != nil {
t.Fatal(err)
}
if uow.states["o-1"] != "PAID" || len(uow.msgs) != 1 {
t.Fatalf("commit: state=%s msgs=%d", uow.states["o-1"], len(uow.msgs))
}
bad := newMemoryUoW("o-1")
bad.failAppend = true
if err := RecordPayment(ctx, bad, "o-1", "pay-123", time.Now()); err == nil {
t.Fatal("want error")
}
if bad.states["o-1"] != "CREATED" || len(bad.msgs) != 0 {
t.Fatalf("rollback: state=%s msgs=%d", bad.states["o-1"], len(bad.msgs))
}
}
type flakyPublisher struct {
calls int
delivered []string
}
func (p *flakyPublisher) Publish(ctx context.Context, m Message) error {
p.calls++
p.delivered = append(p.delivered, m.ID)
if p.calls == 1 {
return errors.New("broker accepted batch but connection dropped")
}
return nil
}
func TestDispatcherAtLeastOnce(t *testing.T) {
ctx := context.Background()
uow := newMemoryUoW("o-1")
if err := RecordPayment(ctx, uow, "o-1", "pay-123", time.Now()); err != nil {
t.Fatal(err)
}
pub := &flakyPublisher{}
d := &Dispatcher{Store: uow, Publisher: pub}
if n, err := d.RunOnce(ctx, 10); err != nil || n != 0 {
t.Fatalf("first run n=%d err=%v", n, err)
}
if n, err := d.RunOnce(ctx, 10); err != nil || n != 1 {
t.Fatalf("second run n=%d err=%v", n, err)
}
if pub.calls != 2 || len(pub.delivered) != 2 || pub.delivered[0] != pub.delivered[1] {
t.Fatalf("calls=%d delivered=%v", pub.calls, pub.delivered)
}
}
func TestDeduper(t *testing.T) {
handled := 0
d := NewDeduper(func(ctx context.Context, m Message) error {
handled++
return nil
})
for i := 0; i < 3; i++ {
if err := d.Handle(context.Background(), Message{ID: "evt-1"}); err != nil {
t.Fatal(err)
}
}
if err := d.Handle(context.Background(), Message{ID: "evt-2"}); err != nil {
t.Fatal(err)
}
if handled != 2 {
t.Fatalf("handler calls = %d, want 2", handled)
}
}
真实运行结果
以下输出来自本机 macOS（darwin/arm64，Go 1.26.0），代码抽取到 .local/checks/ch06-be-arch-base-06-2-transaction-boundary-outbox/ 的独立模块执行：
$ gofmt -l .
$ go vet ./...
$ go test ./... -count=1 -v
=== RUN TestAtomicity
--- PASS: TestAtomicity (0.00s)
=== RUN TestDispatcherAtLeastOnce
--- PASS: TestDispatcherAtLeastOnce (0.00s)
=== RUN TestDeduper
--- PASS: TestDeduper (0.00s)
PASS
ok archlab/internal/outbox 0.495s
gofmt -l . 没有输出，go vet ./... 没有告警。PostgreSQL 表结构没有在本机执行，因此 sent_at 部分索引和唯一约束只作为目标 DDL，需在 L3 用真实执行计划验证。
常见错误与边界
第一，把消息发送放进数据库事务。网络调用不是事务资源，成功与回滚无法同步；事务内只写 outbox，事务外再投递。
第二，先落库再发消息，且没有 outbox。崩溃窗口内消息永久丢失，只能靠对账补偿；对账是最后手段，不是事务边界。
第三，投递成功后不标记 sent_at。消息会被无限重复投递；反过来，先标记后发布则会丢消息。顺序必须是先发布、后标记。
第四，消费端只在内存里去重。进程重启或换实例后去重表清空，重复消息会再次执行业务；生产去重必须持久化，并与业务写入同事务。
小结
还有一个边界：outbox 只解决“本地状态到消息代理”的一段，不解决下游消费失败。消费失败要重试、进 dead-letter 或人工处理，不能回滚已经提交的订单支付；这正是最终一致与本地事务的分界线。
本地事务的边界应该停在数据库内：状态与 outbox 同事务提交，投递器负责把消息送到代理，消费端负责幂等。这个组合不能消除重复，只能把“最多一次”或“可能丢失”换成“至少一次 + 可去重”，这正是最终一致在工程上的可操作形态。6.3 会在这个事件流之上讨论订单读模型：什么时候值得为列表与详情单独建投影，什么时候应该退出。
阅读导航：上一节：6.1 订单数据该放什么存储：所有权与存储选型 ADR · 下一节：6.3 订单读模型：CQRS 的引入判据与退出条件 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。