并发测试框架设计：Fan-in 模式实现预处理与验证的协同调度

2026-01-30 00:00:00 作者：碧海醫心

本文介绍一种基于 go 通道与 waitgroup 的 fan-in 并发模式，用于构建可扩展的集成测试套件——通过预处理器与验证器双工作池协同调度，确保测试按“准备→验证→递归执行子测试”顺序安全并发执行。

在构建高可靠性的集成测试框架时，单纯启动 goroutine 并不足够；关键在于编排依赖关系与收敛异步结果。本方案采用经典的 Fan-in 模式：多个 worker 从共享通道消费任务（Fan-out），每个任务完成后再将结果统一汇入调用方的专属响应通道（Fan-in），从而解耦执行逻辑与控制流。

核心设计解析

双通道分离职责：prepperChan 和 validatorChan 分别承载预处理与验证任务，避免资源争用，也便于独立伸缩并发度（ConcurrentPreppers / ConcurrentValidators）。
Transport 结构体封装任务+回传通道：
```
type prepperTransport struct {
    Prepper Prepper
    Result  chan PrepperResult // 每个测试独享，保证结果精准归属
}
```
此设计替代了全局回调或共享状态，是实现“任务发送即返回结果”的关键——它让每个 Test 能发起一次异步调用，并同步等待其专属结果，语义清晰且线程安全。
递归 + WaitGroup 实现树形测试调度：
runTest 方法中，父测试完成验证后，为每个子测试 Add(1) 并启动 goroutine；子测试自行管理其子树。ct.testSync.Done() 在所有路径（成功、失败、异常）上严格配对，确保 Wait() 精确阻塞至整棵树完成。

完整可运行示例（精简版）

package conctest

import (
    "sync"
    "time"
)

type ConcTest struct {
    Tests                []*Test
    ConcurrentPreppers   int
    ConcurrentValidators int
    prepperChan          chan *prepperTransport
    validatorChan        chan *validatorTransport
    testSync             *sync.WaitGroup
}

func New() *ConcTest {
    return &ConcTest{
        Tests:                nil,
        ConcurrentPreppers:   2,
        ConcurrentValidators: 3,
        prepperChan:          make(chan *prepperTransport),
        validatorChan:        make(chan *validatorTransport),
        testSync:             &sync.WaitGroup{},
    }
}

// 启动工作池（建议在 Run 中调用）
func (ct *ConcTest) startWorkers() {
    for i := 0; i < ct.ConcurrentPreppers; i++ {
        go func() {
            for pt := range ct.prepperChan {
                pt.Result <- pt.Prepper() // 执行并回传
            }
        }()
    }
    for i := 0; i < ct.ConcurrentValidators; i++ {
        go func() {
            for vt := range ct.validatorChan {
                vt.Result <- vt.Validator()
            }
        }()
    }
}

func (ct *ConcTest) Run() {
    ct.startWorkers()
    ct.testSync = &sync.WaitGroup{}

    for _, t := range ct.Tests {
        ct.testSync.Add(1)
        go ct.runTest(t)
    }
    ct.testSync.Wait()
}

func (ct *ConcTest) runTest(t *Test) {
    defer ct.testSync.Done()
    t.Pass = true

    // 预处理：阻塞等待专属结果
    pt := &prepperTransport{t.Prepper, make(chan PrepperResult, 1)}
    ct.prepperChan <- pt
    if err := <-pt.Result; err != nil {
        t.Pass = false
        t.Errors = append(t.Errors, err)
        return
    }

    // 验证循环（含重试）
    for t.Runs < t.MaxRuns {
        t.Runs++
        vt := &validatorTransport{t.Validator, make(chan ValidatorResult, 1)}
        ct.validatorChan <- vt
        vr := <-vt.Result

        if vr.Error != nil {
            t.Errors = append(t.Errors, vr.Error)
        }
        if vr.Pass {
            break
        }
        if t.Runs == t.MaxRuns {
            t.Pass = false
            return
        }
        time.Sleep(t.Frequency)
    }
    if !t.Pass {
        return
    }

    // 递归执行子测试
    for _, child := range t.Children {
        ct.testSync.Add(1)
        go ct.runTest(child)
    }
}