Go 分布式事务方案对比Saga、TCC 和消息最终一致性一、订单创建了但库存没扣——钱付了货没发订单服务调用成功但库存服务超时了。订单已经落库库存却没扣。用户付了款等了两天没收到货客服收到了大量投诉。这就是分布式事务的经典难题跨服务的操作需要要么全成功要么全失败。单体应用里一个BEGIN/COMMIT/ROLLBACK解决的事在微服务里变成了分布式协调问题。三种主流方案Saga补偿事务、TCCTry-Confirm-Cancel、消息最终一致性。二、三种方案的原理与对比flowchart TD subgraph Saga[Saga 模式: 编排/协同] S1[Step1: 创建订单] -- S2[Step2: 扣减库存] S2 -- S3[Step3: 创建支付] S3 -- S4{全部成功?} S4 --|是| S5[完成] S4 --|Step2 失败| S6[补偿: 取消订单] S4 --|Step3 失败| S7[补偿: 取消订单 恢复库存] end subgraph TCC[TCC 模式: 两阶段] T1[Try: 预留资源] -- T2[Confirm: 确认执行] T1 -- T3[Cancel: 释放资源] end subgraph MQ[消息最终一致性] M1[订单服务: 写本地 发消息] -- M2[消息队列] M2 -- M3[库存服务: 消费消息] M3 -- M4{扣减成功?} M4 --|否| M5[重试/死信队列] end方案一致性复杂度适用场景Saga最终一致中长事务、多步骤业务流程TCC强一致(业务层)高资金交易、高一致性要求消息最终一致最终一致低异步解耦、可接受短暂不一致三、Go 实现三种方案的核心代码package disttx import ( context encoding/json fmt log time ) // 方案 1Saga 编排模式 // SagaStep Saga 步骤 type SagaStep struct { Name string Execute func(ctx context.Context, sagaID string) error Compensate func(ctx context.Context, sagaID string) error // 补偿操作 } // SagaOrchestrator Saga 编排器 type SagaOrchestrator struct { steps []SagaStep sagaLog SagaLogger } type SagaState struct { SagaID string Status string // running, completed, compensating, failed StepIndex int StepStates []string // pending, success, failed, compensated } type SagaLogger interface { SaveState(ctx context.Context, state *SagaState) error GetState(ctx context.Context, sagaID string) (*SagaState, error) } func (so *SagaOrchestrator) Execute(ctx context.Context, sagaID string) error { state : SagaState{ SagaID: sagaID, Status: running, StepStates: make([]string, len(so.steps)), } for i, step : range so.steps { state.StepIndex i log.Printf([Saga %s] 执行步骤 %d: %s, sagaID, i1, step.Name) if err : step.Execute(ctx, sagaID); err ! nil { log.Printf([Saga %s] 步骤 %d 失败: %v, sagaID, i1, err) state.Status compensating // 执行补偿回滚已成功的步骤 for j : i - 1; j 0; j-- { log.Printf([Saga %s] 补偿步骤 %d: %s, sagaID, j1, so.steps[j].Name) if err : so.steps[j].Compensate(ctx, sagaID); err ! nil { log.Printf([Saga %s] 补偿失败: %v, sagaID, err) state.Status failed return fmt.Errorf(补偿失败: %w, err) } } state.Status compensated return fmt.Errorf(saga 失败并已补偿: %w, err) } state.StepStates[i] success so.sagaLog.SaveState(ctx, state) } state.Status completed so.sagaLog.SaveState(ctx, state) return nil } // 方案 2TCC 模式 // TCCParticipant TCC 参与者接口 type TCCParticipant interface { Try(ctx context.Context, txID string, data interface{}) error Confirm(ctx context.Context, txID string) error Cancel(ctx context.Context, txID string) error } // TCCCoordinator TCC 协调器 type TCCCoordinator struct { participants []TCCParticipant } func (tc *TCCCoordinator) Execute(ctx context.Context, txID string, data []interface{}) error { // 第一阶段Try预留资源 for i, p : range tc.participants { if err : p.Try(ctx, txID, data[i]); err ! nil { // Try 失败回滚已成功的 Try for j : i - 1; j 0; j-- { tc.participants[j].Cancel(ctx, txID) } return fmt.Errorf(TCC Try 阶段失败: %w, err) } } // 第二阶段Confirm确认提交 for _, p : range tc.participants { if err : p.Confirm(ctx, txID); err ! nil { // Confirm 失败全部 Cancel for _, pp : range tc.participants { pp.Cancel(context.Background(), txID) } return fmt.Errorf(TCC Confirm 阶段失败: %w, err) } } return nil } // 方案 3消息最终一致性 // OutboxMessage 发件箱消息 type OutboxMessage struct { ID string json:id Topic string json:topic Payload string json:payload Status string json:status // pending, sent, failed CreatedAt time.Time json:created_at } // OutboxPublisher 发件箱模式发布者 type OutboxPublisher struct { db OutboxDB mq MessageQueue } type OutboxDB interface { // 在同一个事务中写业务表 写 outbox 表 TransactionalWrite(ctx context.Context, businessFn func(ctx context.Context) error, msg OutboxMessage) error // 查询待发送的消息 FetchPending(ctx context.Context, limit int) ([]OutboxMessage, error) // 标记消息已发送 MarkSent(ctx context.Context, msgID string) error } type MessageQueue interface { Publish(ctx context.Context, topic string, payload []byte) error } // CreateOrderWithOutbox 创建订单使用 Outbox 保证消息一定发出 func (op *OutboxPublisher) CreateOrderWithOutbox(ctx context.Context, orderData interface{}) error { // 序列化事件 payload, _ : json.Marshal(orderData) msg : OutboxMessage{ ID: fmt.Sprintf(msg-%d, time.Now().UnixNano()), Topic: order.created, Payload: string(payload), Status: pending, CreatedAt: time.Now(), } // 在同一个数据库事务中写入订单和消息 return op.db.TransactionalWrite(ctx, func(ctx context.Context) error { // 写订单业务表 // INSERT INTO orders ... return nil }, msg) } // PublishLoop 消息发送循环独立协程运行 func (op *OutboxPublisher) PublishLoop(ctx context.Context) { ticker : time.NewTicker(1 * time.Second) defer ticker.Stop() for { select { case -ctx.Done(): return case -ticker.C: msgs, _ : op.db.FetchPending(ctx, 100) for _, msg : range msgs { if err : op.mq.Publish(ctx, msg.Topic, []byte(msg.Payload)); err ! nil { log.Printf([Outbox] 发送失败: %v, err) continue } op.db.MarkSent(ctx, msg.ID) } } } } // 幂等消费 // IdempotentConsumer 幂等消费者 type IdempotentConsumer struct { processedKeys IdempotentStore } type IdempotentStore interface { HasProcessed(ctx context.Context, msgID string) (bool, error) MarkProcessed(ctx context.Context, msgID string) error } func (ic *IdempotentConsumer) Consume( ctx context.Context, msgID string, handler func(ctx context.Context) error, ) error { // 幂等检查这条消息处理过了吗 processed, _ : ic.processedKeys.HasProcessed(ctx, msgID) if processed { log.Printf([Idempotent] 消息 %s 已处理跳过, msgID) return nil } // 执行业务逻辑 if err : handler(ctx); err ! nil { return err } // 标记已处理 return ic.processedKeys.MarkProcessed(ctx, msgID) }四、选型的边界与常见错误Saga 的补偿不一定完美。如果补偿操作如取消订单也可能失败会导致数据不一致的中间状态。需要重试机制 人工介入流程来处理补偿失败的情况。TCC 的 Try 阶段是性能瓶颈。Try 需要对资源做预留如冻结库存这意味着在高并发场景下可能有大量资源被锁定但不一定被使用。需要在超时后自动释放 Try 锁定的资源。消息最终一致性的最终可能是几分钟。如果消息队列积压或消费者故障最终可能延迟很久。需要监控消息 Lag告警阈值要匹配业务可接受的不一致时长。不是所有场景都需要分布式事务。如果两个操作可以合并在一个服务内完成就不要引入分布式事务的复杂度。能局部化就不分布式化。五、总结分布式事务选型决策一致性和性能始终是权衡。Saga 适合长业务流程的最终一致补偿兜底TCC 适合资金交易场景的强一致Try-Confirm-Cancel 两阶段消息最终一致性适合异步解耦场景Outbox 幂等消费。实施建议从消息最终一致性开始最简单满足不了再上 SagaTCC 只在真正需要强一致性的金融场景使用。