分布式事务实战：从CAP定理到Saga与TCC模式的Go语言实现

4.3 TCC注意事项

• 空回滚防护：Try未收到就收到Cancel时，需要识别并记录，Cancel时直接返回成功
• 幂等性：Confirm和Cancel可能因网络重试被多次调用，需保证幂等
• 悬挂问题：Cancel先于Try执行，Try需检查是否存在已Cancel的记录

五、Saga模式

5.1 工作原理

Saga将一个长事务拆分为一系列本地事务，每个本地事务都有对应的补偿操作。如果某个步骤失败，则按逆序执行已完成步骤的补偿操作。

Saga有两种协调方式：

• 编排式(Choreography)：各服务通过事件驱动，每个服务完成自己的本地事务后发布事件，触发下一个服务
• 协调式(Orchestration)：由中央协调者按顺序调用各服务，失败时触发补偿

5.2 协调式Saga Go实现

package saga

import (
 "context"
 "errors"
 "fmt"
 "time"
)

// SagaStep Saga步骤
type SagaStep struct {
 Name       string
 Action     func(ctx context.Context) error
 Compensate func(ctx context.Context) error
}

// Saga Saga事务定义
type Saga struct {
 Name  string
 Steps []SagaStep
}

func NewSaga(name string) *Saga {
 return &Saga{
  Name:  name,
  Steps: make([]SagaStep, 0),
 }
}

func (s *Saga) AddStep(name string, action func(ctx context.Context) error, compensate func(ctx context.Context) error) {
 s.Steps = append(s.Steps, SagaStep{
  Name:       name,
  Action:     action,
  Compensate: compensate,
 })
}

// Execute 执行Saga事务
func (s *Saga) Execute(ctx context.Context) error {
 completedSteps := make([]int, 0)

 for i, step := range s.Steps {
  stepCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
  fmt.Printf("[Saga] 执行步骤 %d: %s\n", i, step.Name)

  if err := step.Action(stepCtx); err != nil {
   cancel()
   fmt.Printf("[Saga] 步骤 %d 失败: %v，开始补偿\n", i, err)

   // 逆序补偿
   for j := len(completedSteps) - 1; j >= 0; j-- {
    compStep := s.Steps[completedSteps[j]]
    compCtx, compCancel := context.WithTimeout(ctx, 30*time.Second)
    if compErr := compStep.Compensate(compCtx); compErr != nil {
     fmt.Printf("[Saga] 补偿步骤 '%s' 失败: %v（需人工介入）\n", compStep.Name, compErr)
    }
    compCancel()
   }
   return fmt.Errorf("saga '%s' failed at step '%s': %w", s.Name, step.Name, err)
  }

  cancel()
  completedSteps = append(completedSteps, i)
  fmt.Printf("[Saga] 步骤 %d 完成\n", i)
 }

 fmt.Printf("[Saga] 事务 '%s' 全部完成\n", s.Name)
 return nil
}

// 电商下单Saga示例
func OrderSagaExample() {
 ctx := context.Background()
 saga := NewSaga("创建订单")

 var orderID string
 var frozenStock []int
 var deductedPoints int

 // 步骤1：创建订单
 saga.AddStep(
  "创建订单",
  func(ctx context.Context) error {
   orderID = fmt.Sprintf("ORDER_%d", time.Now().Unix())
   fmt.Printf("创建订单: %s\n", orderID)
   return nil
  },
  func(ctx context.Context) error {
   fmt.Printf("取消订单: %s\n", orderID)
   orderID = ""
   return nil
  },
 )

 // 步骤2：扣减库存
 saga.AddStep(
  "扣减库存",
  func(ctx context.Context) error {
   frozenStock = []int{101, 102}
   fmt.Printf("扣减商品库存: %v\n", frozenStock)
   return nil
  },
  func(ctx context.Context) error {
   fmt.Printf("恢复库存: %v\n", frozenStock)
   frozenStock = nil
   return nil
  },
 )

 // 步骤3：扣减积分
 saga.AddStep(
  "扣减积分",
  func(ctx context.Context) error {
   deductedPoints = 100
   fmt.Printf("扣减积分: %d\n", deductedPoints)
   return nil
  },
  func(ctx context.Context) error {
   fmt.Printf("退还积分: %d\n", deductedPoints)
   deductedPoints = 0
   return nil
  },
 )

 // 步骤4：支付
 saga.AddStep(
  "支付",
  func(ctx context.Context) error {
   fmt.Printf("执行支付，订单: %s\n", orderID)
   return nil
  },
  func(ctx context.Context) error {
   fmt.Printf("退款完成\n")
   return nil
  },
 )

 if err := saga.Execute(ctx); err != nil {
  fmt.Printf("下单失败: %v\n", err)
 } else {
  fmt.Println("下单成功！")
 }
}

// 模拟支付失败的Saga
func OrderSagaWithFailureExample() {
 ctx := context.Background()
 saga := NewSaga("创建订单（模拟支付失败）")

 saga.AddStep("创建订单",
  func(ctx context.Context) error { fmt.Println("订单创建成功"); return nil },
  func(ctx context.Context) error { fmt.Println("订单已取消"); return nil },
 )
 saga.AddStep("扣减库存",
  func(ctx context.Context) error { fmt.Println("库存已扣减"); return nil },
  func(ctx context.Context) error { fmt.Println("库存已恢复"); return nil },
 )
 saga.AddStep("积分抵扣",
  func(ctx context.Context) error { fmt.Println("积分已抵扣"); return nil },
  func(ctx context.Context) error { fmt.Println("积分已退还"); return nil },
 )
 saga.AddStep("支付",
  func(ctx context.Context) error { return errors.New("余额不足，支付失败") },
  func(ctx context.Context) error { return nil },
 )

 if err := saga.Execute(ctx); err != nil {
  fmt.Printf("流程终止: %v\n", err)
 }
}

5.3 编排式Saga vs 协调式Saga

六、本地消息表

6.1 工作原理

本地消息表的核心思想是将分布式事务拆分为本地事务，通过保证本地事务和消息投递的一致性来实现最终一致性：

1. 业务执行和消息插入在同一个本地事务中完成
2. 定时任务扫描未发送的消息，投递到消息队列
3. 下游消费者处理消息，成功后回调确认
4. 失败时通过重试机制保证最终送达

6.2 Go实现核心逻辑

package localmsg

import (
 "context"
 "database/sql"
 "fmt"
 "time"
)

// Message 本地消息
type Message struct {
 ID         string
 Topic      string
 Payload    string
 Status     int       // 0:待发送 1:已发送 2:已完成
 RetryCount int
 NextRetry  time.Time
 CreatedAt  time.Time
}

// MessageStore 消息存储接口
type MessageStore interface {
 Save(ctx context.Context, msg *Message, execTx func(tx *sql.Tx) error) error
 ScanPending(ctx context.Context, limit int) ([]*Message, error)
 MarkSent(ctx context.Context, id string) error
 MarkDone(ctx context.Context, id string) error
 IncrementRetry(ctx context.Context, id string) error
}

// DBMessageStore 基于数据库的实现
type DBMessageStore struct {
 db *sql.DB
}

func NewDBMessageStore(db *sql.DB) *DBMessageStore {
 return &DBMessageStore{db: db}
}

// Save 在同一个事务中保存消息和业务数据
func (s *DBMessageStore) Save(ctx context.Context, msg *Message, execTx func(tx *sql.Tx) error) error {
 tx, err := s.db.BeginTx(ctx, nil)
 if err != nil {
  return err
 }
 defer tx.Rollback()

 // 执行业务逻辑
 if err := execTx(tx); err != nil {
  return err
 }

 // 插入消息记录
 _, err = tx.ExecContext(ctx,
  "INSERT INTO local_message (id, topic, payload, status, created_at) VALUES (?, ?, ?, 0, ?)",
  msg.ID, msg.Topic, msg.Payload, time.Now(),
 )
 if err != nil {
  return fmt.Errorf("insert message failed: %w", err)
 }

 return tx.Commit()
}

// ScanPending 扫描待发送消息
func (s *DBMessageStore) ScanPending(ctx context.Context, limit int) ([]*Message, error) {
 rows, err := s.db.QueryContext(ctx,
  "SELECT id, topic, payload, status, retry_count, next_retry FROM local_message WHERE status = 0 AND next_retry <= ? LIMIT ?",
  time.Now(), limit,
 )
 if err != nil {
  return nil, err
 }
 defer rows.Close()

 msgs := make([]*Message, 0)
 for rows.Next() {
  m := &Message{}
  err := rows.Scan(&m.ID, &m.Topic, &m.Payload, &m.Status, &m.RetryCount, &m.NextRetry)
  if err != nil {
   return nil, err
  }
  msgs = append(msgs, m)
 }
 return msgs, nil
}

func (s *DBMessageStore) MarkSent(ctx context.Context, id string) error {
 _, err := s.db.ExecContext(ctx, "UPDATE local_message SET status = 1 WHERE id = ?", id)
 return err
}

func (s *DBMessageStore) MarkDone(ctx context.Context, id string) error {
 _, err := s.db.ExecContext(ctx, "UPDATE local_message SET status = 2 WHERE id = ?", id)
 return err
}

func (s *DBMessageStore) IncrementRetry(ctx context.Context, id string) error {
 _, err := s.db.ExecContext(ctx,
  "UPDATE local_message SET retry_count = retry_count + 1, next_retry = ? WHERE id = ?",
  time.Now().Add(60*time.Second), id,
 )
 return err
}

// MessageWorker 消息投递工作器
type MessageWorker struct {
 store  MessageStore
 mq     MQProducer
 stopCh chan struct{}
}

// MQProducer 消息队列生产者接口
type MQProducer interface {
 Publish(ctx context.Context, topic string, payload []byte) error
}

func NewMessageWorker(store MessageStore, mq MQProducer) *MessageWorker {
 return &MessageWorker{
  store:  store,
  mq:     mq,
  stopCh: make(chan struct{}),
 }
}

func (w *MessageWorker) Start() {
 ticker := time.NewTicker(5 * time.Second)
 defer ticker.Stop()

 for {
  select {
  case <-w.stopCh:
   return
  case <-ticker.C:
   w.process()
  }
 }
}

func (w *MessageWorker) Stop() {
 close(w.stopCh)
}

func (w *MessageWorker) process() {
 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 defer cancel()

 msgs, err := w.store.ScanPending(ctx, 100)
 if err != nil {
  fmt.Printf("扫描消息失败: %v\n", err)
  return
 }

 for _, msg := range msgs {
  if err := w.mq.Publish(ctx, msg.Topic, []byte(msg.Payload)); err != nil {
   fmt.Printf("投递消息 %s 失败: %v\n", msg.ID, err)
   w.store.IncrementRetry(ctx, msg.ID)
   continue
  }
  w.store.MarkSent(ctx, msg.ID)
  fmt.Printf("消息 %s 已投递\n", msg.ID)
 }
}

// 建表SQL
const CreateTableSQL = `CREATE TABLE IF NOT EXISTS local_message (
 id VARCHAR(64) PRIMARY KEY,
 topic VARCHAR(128) NOT NULL,
 payload TEXT NOT NULL,
 status TINYINT NOT NULL DEFAULT 0,
 retry_count INT NOT NULL DEFAULT 0,
 next_retry DATETIME NOT NULL,
 created_at DATETIME NOT NULL,
 INDEX idx_status_next_retry (status, next_retry)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;`

七、方案对比与选型建议

八、最佳实践建议

1. 优先考虑最终一致性：大多数业务场景无需强一致，BASE理论提供的最终一致性已经足够，且能带来更好的性能和可用性
2. TCC用于核心链路：对一致性要求较高的核心链路（如支付、扣款），推荐使用TCC模式
3. Saga应对长流程：涉及3个以上服务的复杂业务流程，使用Saga模式配合补偿机制
4. 本地消息表作为兜底：作为最终手段保证操作消息不丢失，可与上述任何方案配合使用
5. 始终保证幂等性：网络调用不可靠，所有分布式事务操作必须设计为幂等的
6. 记录并监控补偿失败：补偿操作也可能失败，必须记录日志并及时告警，必要时人工介入
7. 超时要短，重试要谨慎：参与者操作的超时时间应远短于全局超时，重试需有上限和退避策略
8. 考虑引入成熟框架：如Seata（支持AT/TCC/SAGA/XA模式）、DTM（Go语言分布式事务管理器）可大幅降低实现成本

九、总结

分布式事务没有银弹，每种方案都有其适用场景和代价。在实际工程中，我们应该：

• 首先判断是否真的需要分布式事务——能通过合理的服务设计规避最好
• 需要时优先选择性能更好的弱一致性方案
• 核心链路使用TCC或Saga + 本地消息表的双重保障
• 建立完善的监控和补偿机制，确保问题可追溯、可修复

CAP定理告诉我们完美不可得，而工程师的艺术在于做出最合理的取舍。希望本文提供的理论分析和Go代码示例，能为你的分布式系统设计提供有价值的参考。