2. Go语言并发基础：goroutine、channel、sync包在事务协调中的应用

分布式事务的本质是跨多个独立资源（数据库、消息队列、服务）的状态协调。在Go语言中，这种协调能力直接依赖于语言内置的并发原语：goroutine、channel 和 sync 包。本章将剖析这三者在事务协调中的具体角色与实战模式。

2.1 并发原语与事务协调的映射关系

事务协调的每个阶段都可以映射到Go的并发模型上。下表展示了核心原语与事务职责的对应关系：

2.2 goroutine：分支事务的轻量执行单元

在SAGA或TCC模式中，每个分支事务（Branch Transaction）都应该运行在独立的 goroutine 中。这样做的好处是：

• 并发执行：多个分支事务可以同时执行，减少整体协调延迟。
• 隔离性：单个分支事务的panic不会影响其他分支。
• 资源轻量：goroutine的栈初始仅几KB，可以轻松创建数千个分支事务协程。

以下是一个典型的SAGA协调器启动分支事务的代码模式：

// SagaCoordinator 启动所有分支事务
func (s *SagaCoordinator) ExecuteBranchTransactions(ctx context.Context, branches []Branch) error {
    errCh := make(chan error, len(branches))
    
    for _, branch := range branches {
        go func(b Branch) {
            // 每个分支事务在独立goroutine中执行
            if err := b.Execute(ctx); err != nil {
                errCh <- err
                return
            }
            errCh <- nil
        }(branch)
    }
    
    // 收集结果（后续结合channel讲解）
    for range branches {
        if err := <-errCh; err != nil {
            return err // 触发补偿流程
        }
    }
    return nil
}

2.3 channel：事务状态的有序传递

channel 是Go中协程间通信的首选方式。在事务协调中，channel 主要用于：

• 结果收集：协调者通过channel接收各分支事务的执行结果。
• 状态广播：协调者通过channel向所有分支广播提交或回滚指令。
• 超时控制：结合 select 和 time.After 实现事务超时。

下图展示了基于channel的事务协调流程：

使用channel实现结果收集的典型代码：

// 使用channel收集分支事务投票结果
func (c *Coordinator) CollectVotes(ctx context.Context, branches []Branch) (bool, error) {
    voteCh := make(chan VoteResult, len(branches))
    
    for _, branch := range branches {
        go func(b Branch) {
            select {
            case <-ctx.Done():
                voteCh <- VoteResult{ID: b.ID, OK: false, Err: ctx.Err()}
            default:
                ok, err := b.Prepare(ctx)
                voteCh <- VoteResult{ID: b.ID, OK: ok, Err: err}
            }
        }(branch)
    }
    
    // 带超时的结果收集
    timeout := time.After(5 * time.Second)
    for i := 0; i < len(branches); i++ {
        select {
        case result := <-voteCh:
            if !result.OK || result.Err != nil {
                return false, result.Err
            }
        case <-timeout:
            return false, ErrTransactionTimeout
        }
    }
    return true, nil
}

2.4 sync.WaitGroup：等待分支事务就绪

在二阶段提交（2PC）或TCC模式的第一阶段，协调者需要等待所有分支事务完成准备操作。 sync.WaitGroup 是等待一组goroutine完成的标准工具。

使用模式：

func (c *Coordinator) WaitForAllBranches(ctx context.Context, branches []Branch) error {
    var wg sync.WaitGroup
    errCh := make(chan error, len(branches))
    
    for _, branch := range branches {
        wg.Add(1)
        go func(b Branch) {
            defer wg.Done()
            if err := b.Prepare(ctx); err != nil {
                errCh <- err
            }
        }(branch)
    }
    
    wg.Wait() // 阻塞直到所有分支完成Prepare
    close(errCh)
    
    // 检查是否有错误
    for err := range errCh {
        if err != nil {
            return err // 触发补偿
        }
    }
    return nil
}

2.5 sync.Mutex：保护共享事务上下文

在事务协调过程中，多个goroutine可能同时访问共享的事务上下文（如事务日志、状态机）。sync.Mutex 用于保护这些临界区。

典型场景——事务日志的并发写入：

type TransactionLog struct {
    mu    sync.Mutex
    entries []LogEntry
}

func (l *TransactionLog) Append(entry LogEntry) {
    l.mu.Lock()
    defer l.mu.Unlock()
    l.entries = append(l.entries, entry)
}

// 在分支事务goroutine中调用
go func() {
    log.Append(LogEntry{
        TxID:      txID,
        BranchID:  branchID,
        Status:    StatusPrepared,
        Timestamp: time.Now(),
    })
}()

2.6 sync.Once：确保幂等提交/回滚

分布式事务中，网络重试可能导致协调者多次发送提交或回滚指令。sync.Once 确保关键操作只执行一次，是实现幂等性的基础。

type BranchTransaction struct {
    commitOnce sync.Once
    rollbackOnce sync.Once
}

func (bt *BranchTransaction) Commit(ctx context.Context) error {
    var err error
    bt.commitOnce.Do(func() {
        err = bt.doCommit(ctx) // 真正的提交逻辑
    })
    return err
}

func (bt *BranchTransaction) Rollback(ctx context.Context) error {
    var err error
    bt.rollbackOnce.Do(func() {
        err = bt.doRollback(ctx) // 真正的回滚逻辑
    })
    return err
}

2.7 综合实战：基于并发原语的SAGA协调器

下面是一个完整的SAGA协调器核心结构，综合运用了上述所有原语：

type SagaCoordinator struct {
    branches   []Branch
    results    chan BranchResult
    wg         sync.WaitGroup
    mu         sync.Mutex
    commitOnce sync.Once
    rollbackOnce sync.Once
    state      SagaState
}

type SagaState int

const (
    StateInit SagaState = iota
    StateExecuting
    StateCompensating
    StateCompleted
    StateFailed
)

func (s *SagaCoordinator) Execute(ctx context.Context) error {
    s.mu.Lock()
    s.state = StateExecuting
    s.mu.Unlock()
    
    // 1. 启动所有分支事务
    s.results = make(chan BranchResult, len(s.branches))
    for _, branch := range s.branches {
        s.wg.Add(1)
        go func(b Branch) {
            defer s.wg.Done()
            err := b.Execute(ctx)
            s.results <- BranchResult{ID: b.ID, Err: err}
        }(branch)
    }
    
    // 2. 等待所有分支完成
    s.wg.Wait()
    close(s.results)
    
    // 3. 检查结果
    for result := range s.results {
        if result.Err != nil {
            s.rollbackOnce.Do(func() {
                s.startCompensation(ctx)
            })
            return result.Err
        }
    }
    
    // 4. 全部成功，提交
    s.commitOnce.Do(func() {
        s.finalizeCommit(ctx)
    })
    return nil
}

func (s *SagaCoordinator) startCompensation(ctx context.Context) {
    s.mu.Lock()
    s.state = StateCompensating
    s.mu.Unlock()
    
    var compWg sync.WaitGroup
    for _, branch := range s.branches {
        compWg.Add(1)
        go func(b Branch) {
            defer compWg.Done()
            b.Compensate(ctx)
        }(branch)
    }
    compWg.Wait()
    
    s.mu.Lock()
    s.state = StateFailed
    s.mu.Unlock()
}

2.8 并发原语选择决策表

在实际开发中，根据不同的协调场景选择合适的并发原语：