Go:关于 Channel-编程知识

文章目录

写在前面
内容
- 模型图与代码
- 发送流程
- 接收流程

写在前面

本篇主要是通过 Channel 的模型图，对 Channel 的原理做一个基本的概述

内容

模型图与代码

我们先来看下 Channel 的模型图：
在这里插入图片描述
以上的图是一个简要的模型图，意味着丢失一些细节，我们再结合源码来看下：

type hchan struct {// 以下是环形缓冲区qcount   uint           // total data in the queuedataqsiz uint           // size of the circular queuebuf      unsafe.Pointer // points to an array of dataqsiz elementselemsize uint16elemtype *_type // element type// Channel 的关闭状态closed   uint32// 以下是等待队列和接收队列sendx    uint   // send indexrecvx    uint   // receive indexrecvq    waitq  // list of recv waiterssendq    waitq  // list of send waiters// 以下是 Channel 的锁lock mutex
}

可以看到，模型图相比于源码，主要是少了 closed 和 lock 这两个属性。
从模型图里我们可以看到，Channel 的构造主要有三部分：

发送队列
接收队列
缓冲区

因此我们经常遇到的问题就出现在这三部分是否有无的排列组合，例如发送队列里有等待协程，接收队列里有等待协程，无缓冲区这样。

接下来我们结合模型图，并分别从发送流程和接收流程来看 Channel 的一个基本运作流程。至于模型图里没有包括的部分（锁和关闭状态）就作为补充处理。
（注：Channel 的源码位于 runtime 包下的 chan.go 文件）

发送流程

发送流程里，主要涉及到的方法是

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool

假设我们现在要发送一个数据，那么这个数据就可能有这么几种状态

在发送等待队列里
在缓冲区里
被接收队列给拿走了

在源码里，发送数据的时候，总体流程上是

我们可以看下：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {...lock(&c.lock)// 向一个已经关闭的通道发送数据，会 panicif c.closed != 0 {unlock(&c.lock)panic(plainError("send on closed channel"))}// 会先检查接收队列里是否有等待接收的协程，如果有等待协程，我们就可以忽略缓冲区的两种情况：// - 没有缓冲区// - 缓冲区满// 那么意味着这里我们的数据是直接跟接收队列对接，既然接收队列里有等待协程，那就把数据给他就行了if sg := c.recvq.dequeue(); sg != nil {send(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}// 到了这一步，就意味着不考虑接收队列了，因为没有等待协程可以用// 于是就看下是否有缓冲区了，有缓冲区的话，就把数据放到缓冲区即可if c.qcount < c.dataqsiz {// Space is available in the channel buffer. Enqueue the element to send.qp := chanbuf(c, c.sendx)if raceenabled {racenotify(c, c.sendx, nil)}typedmemmove(c.elemtype, qp, ep)c.sendx++if c.sendx == c.dataqsiz {c.sendx = 0}c.qcount++unlock(&c.lock)return true}...// 到了这里，说明不考虑接收队列和缓冲区了，那就是要进入发送等待队列了gp := getg()mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}mysg.elem = epmysg.waitlink = nilmysg.g = gpmysg.isSelect = falsemysg.c = cgp.waiting = mysggp.param = nil// 加到发送等待队列里，然后就进入阻塞状态c.sendq.enqueue(mysg)atomic.Store8(&gp.parkingOnChan, 1)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)KeepAlive(ep)// someone woke us up.if mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilgp.activeStackChans = falseclosed := !mysg.successgp.param = nilif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}mysg.c = nilreleaseSudog(mysg)// 再检查一下通道是否关闭了，向一个已经关闭的通道发送数据，会 panicif closed {if c.closed == 0 {throw("chansend: spurious wakeup")}panic(plainError("send on closed channel"))}return true
}

这里面关于 closed 和 lock 有一些细节：

在发送开始的时候，我们需要上锁，等数据要么被拿走，要么进入缓冲区了，要么进入等待队列了，再解锁
- 也就是操作 channel 的发送是需要加锁的
如果一个 channel 被关闭了，此时还向它发送数据，会发生 panic

接收流程

接收数据里，主要涉及的方法是：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool)

那么从接收数据的角度来看，要取一个数据，就会出现几种情况

取到数据
- 从缓冲区取
- 从等待发送队列里取
取不到数据
- 进入等待接收队列

在源码里，它的总体流程是：

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {... lock(&c.lock)// 如果 channel 被关闭if c.closed != 0 {// 如果缓冲区里也没有数据，就直接 return 即可if c.qcount == 0 {... unlock(&c.lock)if ep != nil {typedmemclr(c.elemtype, ep)}return true, false}} else {// 如果 channel 没关闭// 看下等待发送队列里是否有等待协程// 如果有等待发送的协程，那就再去看下缓冲区// 如果没有缓冲区，或是缓冲区没数据，就直接从发送等待队列里拿数据// 如果有缓冲区且有数据，就从缓冲区里拿数据，再把等待队列里的数据追加到缓冲队列的末尾if sg := c.sendq.dequeue(); sg != nil {recv(c, sg, ep, func() { unlock(&c.lock) }, 3)return true, true}}// 到这里就认为不去看接收队列了，直接看缓冲区// 缓冲区有数据，就直接取if c.qcount > 0 {// Receive directly from queueqp := chanbuf(c, c.recvx)if raceenabled {racenotify(c, c.recvx, nil)}if ep != nil {typedmemmove(c.elemtype, ep, qp)}typedmemclr(c.elemtype, qp)c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}c.qcount--unlock(&c.lock)return true, true}...// 想取数据却没地方可取，那就加入接收等待队列，然后就进入阻塞状态了gp := getg()mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.isSelect = falsemysg.c = cgp.param = nilc.recvq.enqueue(mysg)atomic.Store8(&gp.parkingOnChan, 1)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)// someone woke us upif mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilgp.activeStackChans = falseif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}success := mysg.successgp.param = nilmysg.c = nilreleaseSudog(mysg)return true, success
}