Golang实现Barrier栅栏同步多协程方法

Go标准库未提供sync.Barrier，推荐用sync.WaitGroup加channel实现可重用屏障：每个goroutine到达时Add(1)并等待channel，最后一个到达者关闭旧channel、重置计数并创建新channel。

Go 里没有内置 sync.Barrier，得自己造

Go 标准库确实没提供 Barrier 类型，不像 Java 的 CyclicBarrier 或 Python 的 threading.Barrier。这不是遗漏，而是 Go 更倾向用 channel + sync.WaitGroup 组合来表达“所有协程必须等齐再一起往下走”这种语义。硬套其他语言的 Barrier 模型，反而容易写出阻塞、死锁或状态错乱的代码。

常见错误是试图用一个 chan struct{} 让所有 goroutine 往里发信号，再用计数器判断是否齐了——但没人负责关 channel，也没法重用，更没法处理超时或取消。

• 真正可靠的方案是：用 sync.WaitGroup 控制到达登记，用一个 chan struct{}（或 sync.Once 配合闭锁）广播“可以走了”
• 如果需要重复使用（比如多轮 barrier），必须重置计数器，且确保上一轮结束前没人提前进入下一轮
• 别在 barrier 内部做耗时操作，否则会卡住所有等待者

用 sync.WaitGroup + chan struct{} 实现可重用 barrier

这是最贴近原生 Go 风格的做法，清晰、无锁（除 WaitGroup 自身）、支持重用和超时控制。

核心思路：每个 goroutine 到达时调用 wg.Add(1)，然后阻塞在 channel 上；由最后一个到达者（通过原子判断）关闭该 channel，唤醒全部等待者；之后重置 wg 为下一轮准备。

type Barrier struct {
    wg      sync.WaitGroup
    mu      sync.Mutex
    ch      chan struct{}
    size    int
    arrived int32
}
func NewBarrier(n int) *Barrier {
return &Barrier{
ch:   make(chan struct{}),
size: n,
}
}
func (b *Barrier) Await() {
b.wg.Add(1)
if atomic.AddInt32(&b.arrived, 1) == int32(b.size) {
// 最后一个到达：关闭 channel，重置计数
close(b.ch)
b.mu.Lock()
b.arrived = 0
b.mu.Unlock()
b.ch = make(chan struct{}) // 新 channel，支持下一轮
}
<-b.ch // 等待广播
b.wg.Done()
}

注意：ch 必须每次重置为新 channel，否则第二次 <-b.ch 会立刻返回（因已关闭）。也不能用 sync.Once，它不支持重入。

为什么不用 sync.Cond？它太重还容易死锁

有人试过用 sync.Cond + sync.Mutex 实现 barrier，逻辑看似对：每个 goroutine cond.Wait()，最后一个调用 cond.Broadcast()。但问题很实际：

• Cond.Wait() 会自动释放锁，唤醒后需重新获取——如果唤醒时机与下一轮 Await() 交错，可能漏掉广播
• 必须严格配对 Lock/Unlock，goroutine panic 时极易导致锁未释放，整个 barrier 卡死
• Cond 不自带计数，仍需额外变量记录到达数，原子性难保证

相比之下，channel 的关闭行为是 Go 运行时保证的原子操作，<-ch 对已关闭 channel 永远立即返回，天然防重入、防漏唤醒。

真实场景中要注意超时和取消

生产环境的 barrier 几乎都要支持超时或 context 取消，否则一个协程挂掉，整组都会永久阻塞。

改造 Await() 接收 context.Context 是最稳妥的方式，避免用 time.After 单独起 goroutine 去 close channel（竞态风险高）：

func (b *Barrier) AwaitCtx(ctx context.Context) error {
    b.wg.Add(1)
    done := make(chan error, 1)
    go func() {
        if atomic.AddInt32(&b.arrived, 1) == int32(b.size) {
            close(b.ch)
            b.mu.Lock()
            b.arrived = 0
            b.mu.Unlock()
            b.ch = make(chan struct{})
        }
        done <- nil
    }()
    select {
    case <-ctx.Done():
        return ctx.Err()
    case err := <-done:
        if err != nil {
            return err
        }
    }
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-b.ch:
        b.wg.Done()
        return nil
    }
}

这里的关键点是：登记到达和触发广播必须在同一个 goroutine 内完成，且不能被 ctx 中断——否则计数会错。真正的等待阶段才受 ctx 控制。

复杂点在于，你得确保所有 goroutine 都调用了 AwaitCtx，且传入的是同一个 context.WithTimeout；如果混用不同 context，超时行为就不可预测。这点很容易被忽略。