前言

channels 是一种类型安全的消息队列，充当两个 goroutine 之间的管道，将通过它同步的进行任意资源的交换。chan 控制 goroutines 交互的能力从而创建了 Go 同步机制。当创建的 chan 没有容量时，称为无缓冲通道。反过来，使用容量创建的 chan 称为缓冲通道。 要了解通过 chan 交互的 goroutine 的同步行为是什么，我们需要知道通道的类型和状态。根据我们使用的是无缓冲通道还是缓冲通道，场景会有所不同，所以让我们单独讨论每个场景。

---

使用技巧

Channel读写特性(15字口诀)

首先，我们先复习一下Channel都有哪些特性？

• 给一个 nil channel 发送数据，造成永远阻塞

• 从一个 nil channel 接收数据，造成永远阻塞

• 给一个已经关闭的 channel 发送数据，引起 panic

• 从一个已经关闭的 channel 接收数据，如果缓冲区中为空，则返回一个零值

• 无缓冲的channel是同步的，而有缓冲的channel是非同步的

以上5个特性是死东西，也可以通过口诀来记忆：“空读写阻塞，写关闭异常，读关闭空零”。

channel 的使用

Unbuffered Channels --无缓冲的通道

无缓冲 chan 没有容量，因此进行任何交换前需要两个 goroutine 同时准备好。当 goroutine 试图将一个资源发送到一个无缓冲的通道并且没有goroutine 等待接收该资源时，该通道将锁住发送 goroutine 并使其等待。当 goroutine 尝试从无缓冲通道接收，并且没有 goroutine 等待发送资源时，该通道将锁住接收 goroutine 并使其等待。

 同步是发送和接收之间的交互所固有的。没有另一个就不可能发生。非缓冲信道的本质是保证同步 

例子

	ch := make(chan int)var wg  sync.WaitGroupwg.Add(2)go func() {defer wg.Done()ch <- 1}()go func() {defer wg.Done()id := <-chfmt.Println(id)}()wg.Wait()

g2 必须等待g1 向chen 发送的数据 两者处于同步

一个 goroutine 在发送消息 foo 之后被阻塞，因为还没有接收者准备好。规范中对这种行为进行了很好的解释：

https://golang.org/ref/spec#Channel_types

If the capacity is zero or absent, the channel is unbuffered and communication succeeds only when both a sender and receiver are ready.

如果容量为零或不存在，则信道不缓冲，通信只有在发送方和接收方都准备好时才能成功。

https://golang.org/doc/effective_go.html#channels

If the channel is unbuffered, the sender blocks until the receiver has received the value

如果信道未缓冲，则发送方将阻塞，直到接收方接收到该值为止

Receive 先于 Send 发生。

好处: 100% 保证能收到。

代价: 延迟时间未知。

Buffered Channels

缓冲通道具有容量，因此可以表现得有点不同。当 goroutine 尝试将资源发送到缓冲通道并且该通道已满时，该通道将锁定 goroutine 并使其等待，直到缓冲区变为可用。如果通道中有空间，发送就可以立即进行，通道就可以继续前进。当 goroutine 尝试从缓冲通道接收，缓冲通道为空时，该通道将锁定 goroutine 并使其等待，直到资源被发送

ch := make(chan int,2)
var wg  sync.WaitGroupwg.Add(2)
go func() {defer wg.Done()ch <- 1ch <- 2
}()
go func() {defer wg.Done()id := <-ch  id1 := <-chfmt.Println(id,id1)
}()wg.Wait()

由于缓冲区大小不足而造成的延迟

不是缓存区越大越好 ，看下面的例子

//放弃
func main()  {const cap = 5ch := make(chan int,cap)go func() {for p := range ch{fmt.Println("接受到：",p)}}()const work = 20for i := 0; i < work; i++ {select {case ch <- i:fmt.Println("传入--",i)default:fmt.Println("drep")}}}

如果缓存区满了，在次写入的数据就会丢失造成影响

回归正传 看下面的例子那个处理的时间短呢？

 

package benchimport ("sync""sync/atomic""testing"
)func BenchmarkWithNoBuffer(b *testing.B) {benchmarkWithBuffer(b, 0)
}func BenchmarkWithBufferSizeOf1(b *testing.B) {benchmarkWithBuffer(b, 1)
}func BenchmarkWithBufferSizeEqualsToNumberOfWorker(b *testing.B) {benchmarkWithBuffer(b, 5)
}func BenchmarkWithBufferSizeExceedsNumberOfWorker(b *testing.B) {benchmarkWithBuffer(b, 25)
}func benchmarkWithBuffer(b *testing.B, size int) {for i := 0; i < b.N; i++ {c := make(chan uint32, size)var wg sync.WaitGroupwg.Add(1)go func() {defer wg.Done()for i := uint32(0); i < 1000; i++ {c <- i%2}close(c)}()var total uint32for w := 0; w < 5; w++ {wg.Add(1)go func() {defer wg.Done()for {v, ok := <-cif !ok {break}atomic.AddUint32(&total, v)}}()}wg.Wait()}
}

goos: windows
goarch: amd64
pkg: 03-go-concurrency/sync_test/07
BenchmarkWithNoBuffer
BenchmarkWithNoBuffer-6                                     4297            3527
02 ns/op
BenchmarkWithBufferSizeOf1
BenchmarkWithBufferSizeOf1-6                                4297            2780
72 ns/op
BenchmarkWithBufferSizeEqualsToNumberOfWorker
BenchmarkWithBufferSizeEqualsToNumberOfWorker-6             6015            2104
72 ns/op
BenchmarkWithBufferSizeExceedsNumberOfWorker
BenchmarkWithBufferSizeExceedsNumberOfWorker-6              8594            1412
05 ns/op
PASS

一个适当大小的缓冲区确实可以使您的应用程序更快！让我们分析基准测试的跟踪，以确定延迟的位置。

Send 先于 Receive 发生。

好处: 延迟更小。

代价: 不保证数据到达，越大的 buffer，越小的保障到达。buffer = 1 时，给你延迟一个消息的保障。

三 例子

Timing out

timeout := make(chan bool, 1)
ch := make(chan int, 1)
go func() {time.Sleep(1 * time.Second)timeout <- true
}()
select {
case <-ch:// a read from ch has occurred
case <-timeout:// the read from ch has timed out
}

一个很好的例子就是超时。虽然 Go 的通道不直接支持它们，但是它们很容易实现。说我们想从通道接收 ch，但是最多要等一秒钟才能到值。我们将首先创建一个信令通道，并启动一个在通过该通道发送信号之前处于睡眠状态的通道

然后，我们可以使用 select 语句从 ch 或者 timeout 接收。如果一秒钟后没有任何内容到达 ch，则选择超时案例，并放弃从 ch 读取的尝试。

超时通道用1值的空间缓冲，允许超时通道发送到通道，然后退出。Goroutine 不知道(或关心)是否接收到值。这意味着，如果 ch 接收发生在超时之前，goroutine 不会永远留在这里。超时通道最终将由垃圾回收器释放。

woker 打工人模式

import ("errors""fmt""runtime""sync""sync/atomic"
)//接口
type PoolWorker interface {DoWork(WorkRoutine int)
}//poolWork被传递到要执行的工作的队列中。
type Poolwork struct {work          PoolWorker //需要执行的任务resultChannel chan error //任务返回信息
}var ErrCapacity =  errors.New("Thread Pool At Capacity")type WorkPool struct {shutdownQueueChannel chan string     // 关闭队列的通道shutdownWorkChannel  chan struct{}   // 关闭工作的通道shutdownWaitGroup    sync.WaitGroup  //用于关闭现有的workqueueChannel         chan Poolwork   // 用于同步对队列的访问的通道。workChannel          chan PoolWorker //用于工作的通道queuedWorkNum        int32           // 排队的工作数activeRoutinesNum    int32           //活动的work 数量queueCapacityNum     int32           // 队列中可存储的最大项目数。
}func New(WorkRoutinesNum int,queueCapacityNum int32)  *WorkPool {workPool := WorkPool{shutdownQueueChannel: make(chan string),shutdownWorkChannel: make(chan struct{}),queueChannel: make(chan Poolwork),workChannel: make(chan PoolWorker),queuedWorkNum: 0,activeRoutinesNum: 0,queueCapacityNum: queueCapacityNum,}//开启 groupworkPool.shutdownWaitGroup.Add(WorkRoutinesNum)//启动workfor i := 0; i < WorkRoutinesNum; i++ {go workPool.workRoutine(i)}//启动queuego workPool.queueRoutine()return &workPool}//停止
func (this *WorkPool) Shutdown() (err error) {//捕获错误defer func() {if r := recover(); r != nil {buf := make([]byte, 10000)runtime.Stack(buf, false)fmt.Println(string(buf))err = fmt.Errorf("%v", r)}}()fmt.Println("Shutdown : Queue Routine")//通知 接受任务的协程停止this.shutdownQueueChannel <- "down"//接受  接受任务的协程返回后<-this.shutdownQueueChannel//关闭shutdownQueueChannelclose(this.shutdownQueueChannel)//接受任务close(this.queueChannel)//关闭工作 Routineclose(this.shutdownWorkChannel)//等待工作协程减少this.shutdownWaitGroup.Wait()//关闭工作chenclose(this.workChannel)return  err
}//work
func (this *WorkPool) workRoutine(workRoutineId int)  {for  {select {case <-this.shutdownWorkChannel:  //接受关闭fmt.Println("workRoutine : ",workRoutineId,"正在结束")this.shutdownWaitGroup.Done() //减少协程returncase item := <- this.workChannel: //接受任务this.doWork(item,workRoutineId) break}}}//执行任务
func (this *WorkPool) doWork(work PoolWorker,workRoutineId int)  {defer fmt.Println("执行完成")defer atomic.AddInt32(&this.activeRoutinesNum,-1)  //工作协程 减少atomic.AddInt32(&this.activeRoutinesNum,1)   //工作协程 增加atomic.AddInt32(&this.queuedWorkNum,-1)      //等待任务减少work.DoWork(workRoutineId)
}//添加任务
func (this *WorkPool) PostWork (goRoutine string,worker PoolWorker) (err error) {defer fmt.Println("goRoutine err: ",err)poolwork := Poolwork{work: worker,resultChannel: make(chan error)}this.queueChannel <- poolwork //将任务添加到 任务chenerr = <- poolwork.resultChannel  //同步等待错误close(poolwork.resultChannel)  //关闭结果 chenreturn err //返回错误
}func (this *WorkPool) queueRoutine()  {for  {select {case <-this.shutdownQueueChannel:  //接受任务chen 终止信号fmt.Println("终止接受任务")    this.shutdownQueueChannel <- "down"  //回复收到终止信号returncase queueItem := <- this.queueChannel:   //同步接受任务//是否已达到处理任务   队列中可存储的最大项目数。if atomic.AddInt32(&this.queuedWorkNum,0) == this.queueCapacityNum{queueItem.resultChannel <- ErrCapacitybreak}atomic.AddInt32(&this.queuedWorkNum,1) //增加存储的最大项目数this.workChannel <- queueItem.work  //向工作chen 投递queueItem.resultChannel <- nil  // 返回nilbreak}}}

下面是调用的逻辑 注意 工作协程 请不要开启太多

type MyTask struct {Name stringOrderId intWP *workpool.WorkPool
}func (mt *MyTask) DoWork(workRoutine int) {fmt.Println(mt.Name)fmt.Printf("*******> WR: %d QW : %d AR: %d\n",workRoutine,1,2)fmt.Println(mt.OrderId)time.Sleep(100 * time.Millisecond)
}func main()  {runtime.GOMAXPROCS(runtime .NumCPU())workPool := workpool.New(runtime.NumCPU() * 3, 100)task := MyTask{Name: "A" + strconv.Itoa(1),OrderId : 1,WP: workPool,}err := workPool.PostWork("main",&task)workPool.Shutdown()fmt.Println(err)
}

总结

在使用通道（或并发）时，要了解和理解围绕保证、通道状态和发送的信令属性很重要。它们将帮助指导您实现您正在编写的并发程序和算法所需的最佳行为。他们将帮助您找到错误并嗅出潜在的错误代码。

stages close their outbound channels when all the send operations are done.

当所有的发送操作完成后，stages 关闭它们的出站通道。

stages keep receiving values from inbound channels until those channels are closed or the senders are unblocked.

阶段保持接收来自入站通道的值，直到这些通道关闭或发件人解除阻塞。

Pipelines unblock senders either by ensuring there’s enough buffer for all the values that are sent or by explicitly signalling senders when the receiver may abandon the channel.

通过确保有足够的缓冲区容纳所有发送的值，或者在接收方可能放弃信道时显式发送方信号，管道可以解除对发送方的阻塞

• 使用通道来编排和协调 goroutine。
  • 关注通道属性而不是数据共享。
  • 有数据或无数据的信令。
  • 质疑它们用于同步访问共享状态的用途。
    • 在某些情况下，渠道可以更简单，但最初是疑问。
• 无缓冲通道：
  • 接收发生在发送之前。
  • 优点：100% 保证信号已被接收。
  • 成本：接收信号时的未知延迟。
• 缓冲通道：
  • 发送发生在接收之前。
  • 好处：减少信号之间的阻塞延迟。
  • 成本：无法保证何时收到信号。
    • 缓冲区越大，保证越少。
    • 缓冲区为 1 可以为您提供一次延迟发送保证。

设计理念

• 如果通道上的任何给定 Send 可以导致发送 goroutine 阻塞：
  • 不允许使用大于 1 的缓冲通道。
    • 大于 1 的缓冲区必须有原因/度量。
  • 必须知道发送 goroutine 阻塞时会发生什么。
• 如果通道上的任何给定 Send 不会导致发送 goroutine 阻塞：
  • 您有每次发送的确切缓冲区数。
    • 扇出模式
  • 您测量了最大容量的缓冲区。
    • 掉落图案
• 缓冲区少即是多。
  • 在考虑缓冲区时不要考虑性能。
  • 缓冲区可以帮助减少信号之间的阻塞延迟。
    • 将阻塞延迟降低到零并不一定意味着更好的吞吐量。
    • 如果一个缓冲区为您提供足够好的吞吐量，请保留它。
    • 问题缓冲区大于 1 并测量大小。
    • 找到可能提供足够好的吞吐量的最小缓冲区。

文献

https://medium.com/a-journey-with-go/go-buffered-and-unbuffered-channels-29a107c00268

https://www.ardanlabs.com/blog/2014/02/the-nature-of-channels-in-go.htm

https://www.ardanlabs.com/blog/2013/10/my-channel-select-bug.html

https://blog.golang.org/concurrency-timeouts

https://blog.golang.org/pipelines

https://talks.golang.org/2013/advconc.slide#1

https://github.com/go-kratos/kratos/tree/master/pkg/sync