写这个博客的背景是我面试一家公司，这家公司的CTO给我出了一道我认为挺有意思的题，题的大概是这样的：

// 抽象一个栅栏
type Barrier interface {
    Wait ()
}
// 创建栅栏对象
func NewBarrier (n int) Barrier {
   
}
// 栅栏的实现类
type barrier struct {

}
// 测试代码
func main () {
    // 创建栅栏对象
    b := NewBarrier(10)
    // 达到的效果：前9个协程调用Wait()阻塞，第10个调用后10个协程全部唤醒
    for i:=0; i<10; i++ {
        go b.Wait()
    }
}

需要对上面的NewBarrier()函数和barrier这个类进行修改，达到预期的效果。而且还要有条件约束，就是不能用任何同步相关的操作，但可以用chan，前提是无缓冲模式的。

在做这个问题的时候我一直在想C里面有人实现无锁队列，利用的CPU的CAS指令可以不用加锁就可以对变量实现原子操作。先不说golang中是不是支持此类的语言，其实原子操作也会是同步操作的范畴，本身就已经犯规了，所以还是老老实实的用chan是实现吧。对chan有了解的都知道，没有缓冲的生产和消费必须同时调用二者都会被唤醒，否则任意一方都会被阻塞。至少调用Wait()阻塞的方法有一个选择了，那我们的先可以这么实现：

type barrier struct {
    chCount chan struct{} // 所有调用Wait()函数的先通过这个chan阻塞
    count   int           // 记住数量，阻塞超过这个量就可以激活所有协程了
}
// 按照上面的定义，构造函数就可以这么写了
func NewBarrier(n int) Barrier {
    b := &barrier{count: n, chCount: make(chan struct{}))}
}
// Wait()可以先写成这样
func (b *barrier) Wait() {
    b.chCount <- struct{}{}
}

上面的代码至少实现了所有的协程都能阻塞了，那么问题来了，我通过什么方式计数呢？因为chan没有缓冲，没法用cap()函数获取数量。如果定义全局变量的方式计数，没有锁或者原子操作是没法正确统计计数的，唯一的方式就是从chan中一个一个的读出来计数。那么就会增加如下代码：

func NewBarrier(n int) Barrier {
    b := &barrier{count: n, chCount: make(chan struct{}), chSync: make(chan struct{})}
    go b.Sync() // 增加一个协程用来后台计数
    return b
}
// 后台计数的协程
func (b *barrier) Sync() {
    count := 0
    // chan也可以通过range 操作的，感兴趣的同学可以看我的《深入浅出golang之chan》
    for range b.chCount {
        // 累计计数
        count++
        if count >= b.count {
            // 这里就是统计满足条件的地方了
        }
    }
}

现在我们面临了新的问题，当后台这个协程每次从chan读取一个元素的时候，那个发送该数据阻塞的协程就会被唤醒，这个就没法满足我们的要求了。有什么方法能让发送协程不被激活么,如果是当前状态的chan是无解的。那只能再想一个方法就是让那个写入协程再次阻塞，起初我的想法可以用Sleep()函数，后台协程累计计数达到条件后设置一个标记，写入协程通过循环Sleep()一段时间判断这个标记就可以搞定了。但是作为老司机的我根本没法接受这么丑陋的代码，我再用一个chan阻塞不就可以了么？当统计计数满足条件，直接close掉这个chan，所有的协程就自动激活了，所以代码就变成了这样：

type barrier struct {
    chCount chan struct{}
    chSync  chan struct{}     // 增加一个chan
    count   int
}
func NewBarrier(n int) Barrier {
    b := &barrier{count: n, chCount: make(chan struct{}), chSync: make(chan struct{})}
    go b.Sync()
    return b
}
func (b *barrier) Wait() {
    b.chCount <- struct{}{}
    <-b.chSync                // 再次阻塞
}
func (b *barrier) Sync() {
    count := 0
    for range b.chCount {
        count++
        if count >= b.count {
            close(b.chSync)   // close这个chan所有阻塞协程都会被激活
            break
        }
    }
}

完美解决问题，如果仅仅是解决这个问题我就不会写这个文章了。在这个事情上我想到了两个chan可以实现很多很有意思的功能，上面提到的问题就是其中之一。我再说一个例子：设计一个分发器，当有数据进入分发器后，需要将数据分发到多个处理器处理，每个处理器可以想象为一个协程，处理器在没有数据的时候要阻塞。我相信很多人肯定会设计成这样：

这种设计的最大的问题在于多个Processor处理时长不同会造成木桶效应，多个Processor会被一个Processor拖累。那有人肯定会说可以给chan加缓冲啊，试问缓冲设计多大合适呢？如果真存在一个处理非常慢的Processor多大的缓冲都无济于事，所以应该设计成这样：

有人肯定会说这个chan带buffer没啥区别啊，我告诉你这区别大了！用一个专用的协程实现从chan2读取数据放到缓冲中，然后再冲缓冲中读取数据放到chan3中，全程chan2和chan3都无需缓冲。这样分发器不会由于任何一个Processor慢被拖累，同时缓冲Buffer可以设计成弹性的Buffer，不会被设定成一个固定的值。还有谁有非常有趣的关于chan使用，都可以提出来大家一起分享！