go标准库不支持分布式事务,因缺乏2pc/saga协调器,database/sql仅支持单机事务;需下沉至业务层,用saga模式通过幂等操作、状态持久化和异步补偿保障最终一致性。

为什么不能直接用 Go 标准库做分布式事务
Go 语言标准库没有内置的两阶段提交(2PC)或 Saga 协调器,database/sql 的 BeginTx 只管单机事务,跨服务、跨数据库时会立刻失效。你调用 A 服务扣库存成功,B 服务记账失败,rollback 对 B 已发生的 HTTP 请求毫无作用——这时候不是“要不要补偿”,而是“不补偿就数据不一致”。
真正要做的,是把“事务”从数据库下沉到业务逻辑层,靠可重试 + 幂等 + 显式状态机来兜底。
用 Saga 模式实现补偿的关键三要素
Saga 不是框架,而是一种编排思路:每个操作配一个反向操作,失败时按执行逆序逐个调用补偿函数。
使用 uber-go/dig 在 Golang 中实现依赖注入——基于反射容器,提供 Provide/Invoke、dig.In/dig.Out 参数与结果对象,支持命名依赖等
-
正向操作和补偿操作必须成对定义,且都具备幂等性(比如用order_id+step_id做唯一键插入或UPDATE ... WHERE status = 'pending') - 必须持久化每一步的执行状态(建议用一张
saga_records表,字段至少含:trace_id、step、status(pending/compensated/succeeded)、created_at、updated_at) - 补偿触发不能只靠“上一步失败 panic”,要支持异步扫描+重试(避免阻塞主链路),例如用定时任务查
status = 'pending'且updated_at
// 示例:转账 Saga 的两个步骤
func TransferOut(ctx context.Context, tx *sql.Tx, orderId string, amount int) error {
_, err := tx.ExecContext(ctx, "UPDATE accounts SET balance = balance - ? WHERE user_id = ? AND balance >= ?", amount, "A", amount)
return err
}
func CompensateTransferOut(ctx context.Context, tx *sql.Tx, orderId string, amount int) error {
_, err := tx.ExecContext(ctx, "UPDATE accounts SET balance = balance + ? WHERE user_id = ?", amount, "A")
return err
}
如何保证补偿操作本身不丢、不重复
补偿不是“尽力而为”,而是“必须生效”。常见错误是补偿函数里直接发 HTTP 请求后就返回,结果请求超时或对方没收到,整个流程卡死。
- 补偿动作本身要先落库:往
saga_records插入一条step = 'compensate_transfer_out'、status = 'pending'的记录 - 再用独立 worker 轮询该表,对
status = 'pending'的记录发起真实补偿请求;成功后UPDATE ... SET status = 'compensated' - HTTP 请求必须带超时(
context.WithTimeout(ctx, 8*time.Second))和重试(最多 3 次,每次间隔指数退避),失败仍保持pending等下次扫描 - 所有 DB 操作和 HTTP 调用之间**不能共用同一个
sql.Tx** —— 补偿是跨服务的,只能靠最终一致性,不是 ACID
Go 里最轻量但够用的补偿调度结构
不需要引入 Temporal 或 Cadence 这类重型引擎。一个基于 time.Ticker + sync.Mutex 的本地轮询器,在中小流量下完全够用:
func (s *SagaScheduler) startCompensationLoop() {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for range ticker.C {
s.compensatePendingSteps()
}
}
<p>func (s *SagaScheduler) compensatePendingSteps() {
rows, err := s.db.Query("SELECT trace_id, step, payload FROM saga_records WHERE status = 'pending' ORDER BY created_at LIMIT 100")
if err != nil { return }
for rows.Next() {
var traceID, step, payload string
if err := rows.Scan(&traceID, &step, &payload); err != nil { continue }
if err := s.executeCompensation(step, payload); err == nil {
s.db.Exec("UPDATE saga_records SET status = 'compensated' WHERE trace_id = ? AND step = ?", traceID, step)
}
}
}</p>
注意:这里没加分布式锁,所以如果部署多实例,得用 SELECT ... FOR UPDATE 或 Redis 分布式锁来避免重复处理同一笔 pending 记录——这点容易被忽略,一上生产就双倍补偿。
golang免费学习笔记(深入):立即使用
在学习笔记中,你将探索golang的核心概念和高级技巧!










