文章目录
- 写在前面
- 内容
- 模型图与代码
- 发送流程
- 接收流程
写在前面
本篇主要是通过 Channel 的模型图，对 Channel 的原理做一个基本的概述
内容
模型图与代码
我们先来看下 Channel 的模型图：
以上的图是一个简要的模型图，意味着丢失一些细节，我们再结合源码来看下：
type hchan struct {
// 以下是环形缓冲区
qcount uint // total data in the queue
dataqsiz uint // size of the circular queue
buf unsafe.Pointer // points to an array of dataqsiz elements
elemsize uint16
elemtype *_type // element type
// Channel 的关闭状态
closed uint32
// 以下是等待队列和接收队列
sendx uint // send index
recvx uint // receive index
recvq waitq // list of recv waiters
sendq 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)
// 向一个已经关闭的通道发送数据，会 panic
if 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 = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.waiting = mysg
gp.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 = nil
gp.activeStackChans = false
closed := !mysg.success
gp.param = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
mysg.c = nil
releaseSudog(mysg)
// 再检查一下通道是否关闭了，向一个已经关闭的通道发送数据，会 panic
if 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 queue
qp := 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 = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
gp.waiting = mysg
mysg.g = gp
mysg.isSelect = false
mysg.c = c
gp.param = nil
c.recvq.enqueue(mysg)
atomic.Store8(&gp.parkingOnChan, 1)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
// someone woke us up
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
gp.activeStackChans = false
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
success := mysg.success
gp.param = nil
mysg.c = nil
releaseSudog(mysg)
return true, success
}
同样的，接收的过程里，也有 closed 和 lock 的一些细节：
- 在接收过程前，需要加锁，处理完后再解锁
- 从一个已经关闭的 channel 里取数据，不会造成 panic
