Golang之channel

前言

go中的一个精髓就是就是channel,那么你有没有想过,它究竟是怎么实现的呢?我之前就怀疑过,是不是就是通过一个数组保存了一下传入的数据,然后在接收方读一读就完事了,那么阻塞又是怎么实现的呢?close的时候需要注意些什么呢?

结构

首先我们来看一下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
 closed uint32
 elemtype *_type // element type
 sendx uint // send index
 recvx uint // receive index
 recvq waitq // list of recv waiters
 sendq waitq // list of send waiters
 // lock protects all fields in hchan, as well as several
 // fields in sudogs blocked on this channel.
 //
 // Do not change another G's status while holding this lock
 // (in particular, do not ready a G), as this can deadlock
 // with stack shrinking.
 lock mutex
}

其实看注释这几个字段都非常好理解,解释一下其中几个:
elemtype是表示这个channel中存放的是什么类型的数据;
sendx、recvx两个索引指向底层循环数组
recvq、sendq两个双向链表保存那些等待的goroutine
lock?对就是lock,不然你以为并发的时候channel怎么办?锁呗。

PS: 其实和我一开始想的差不多,底层就是利用一个循环数组来实现的带有缓冲的channel,利用两个index标记的移动来记录发送和读取,然后用一个计数器表示当前还有多少个元素,easy

但是如果你想着go只有这么点东西,那你就太小看它了,细节能把你看哭,嘿嘿嘿,下面来看看源码中具体的接收和发送是怎么实现的。

实现

本质:channel发送接收数据的本质是数据拷贝!

接收

我会删除其中一些细节部分,留下其中重要的点看一下,如果希望看到全部,请自行阅读源码。

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
 // ......
 // 如果收一个nil的channel不会panic的,而是被阻塞,gopark就是将当前goroutine阻塞
 if c == nil {
 if !block {
 return
 }
 gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)
 throw("unreachable")
 }
 // ......
 // 加锁哦!防止并操作channel
 lock(&c.lock)
 // 处理关闭的情况和无data的情况
 if c.closed != 0 && c.qcount == 0 {
 if raceenabled {
 raceacquire(unsafe.Pointer(c))
 }
 unlock(&c.lock)
 if ep != nil {
 typedmemclr(c.elemtype, ep)
 }
 return true, false
 }
 // 当无缓冲 或者 是有缓冲但是缓冲满了 这两种情况下去recv
 if sg := c.sendq.dequeue(); sg != nil {
 recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
 return true, true
 }
 
 // 剩下的情况就是有缓冲的情况,如果有数据的话进if里面,里面其实就是将缓冲中的数据拿出来,并且移动相对应的索引,减少qcount
 if c.qcount > 0 {
 // Receive directly from queue
 qp := chanbuf(c, c.recvx)
 if raceenabled {
 raceacquire(qp)
 racerelease(qp)
 }
 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
 }
 if !block {
 unlock(&c.lock)
 return false, false
 }
 // 如果当前没有数据,那么只能阻塞咯
 // no sender available: block on this channel.
 gp := getg()
 mysg := acquireSudog()
 mysg.releasetime = 0
 if t0 != 0 {
 mysg.releasetime = -1
 }
 // No stack splits between assigning elem and enqueuing mysg
 // on gp.waiting where copystack can find it.
 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)
 goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
 // someone woke us up
 if mysg != gp.waiting {
 throw("G waiting list is corrupted")
 }
 gp.waiting = nil
 if mysg.releasetime > 0 {
 blockevent(mysg.releasetime-t0, 2)
 }
 closed := gp.param == nil
 gp.param = nil
 mysg.c = nil
 releaseSudog(mysg)
 return true, !closed
}

发送

发送其实和接受异曲同工,也是处理其中几种情况

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
 // 发给一个nil的channel就panic呗
 if c == nil {
 if !block {
 return false
 }
 gopark(nil, nil, "chan send (nil chan)", traceEvGoStop, 2)
 throw("unreachable")
 }
 // ......
 // 如果
 // 不是缓冲的channel而且没有接受者正在接受
 // 是缓冲的channel但是缓冲满了
 // 那就直接返回
 if !block && c.closed == 0 && ((c.dataqsiz == 0 && c.recvq.first == nil) ||
 (c.dataqsiz > 0 && c.qcount == c.dataqsiz)) {
 return false
 }
 // ......
 // 加锁!
 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 {
 raceacquire(qp)
 racerelease(qp)
 }
 typedmemmove(c.elemtype, qp, ep)
 c.sendx++
 if c.sendx == c.dataqsiz {
 c.sendx = 0
 }
 c.qcount++
 unlock(&c.lock)
 return true
 }
 if !block {
 unlock(&c.lock)
 return false
 }
 // 当没有缓冲了,那么就需要阻塞发送人了
 gp := getg()
 mysg := acquireSudog()
 mysg.releasetime = 0
 if t0 != 0 {
 mysg.releasetime = -1
 }
 // No stack splits between assigning elem and enqueuing mysg
 // on gp.waiting where copystack can find it.
 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)
 goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
 // someone woke us up.
 if mysg != gp.waiting {
 throw("G waiting list is corrupted")
 }
 gp.waiting = nil
 if gp.param == nil {
 if c.closed == 0 {
 throw("chansend: spurious wakeup")
 }
 panic(plainError("send on closed channel"))
 }
 gp.param = nil
 if mysg.releasetime > 0 {
 blockevent(mysg.releasetime-t0, 2)
 }
 mysg.c = nil
 releaseSudog(mysg)
 return true
}

关闭

func closechan(c *hchan) {
 // 关闭一个nil的channel 那就panic
 if c == nil {
 panic(plainError("close of nil channel"))
 }
 // 关闭也是要加锁的!
 lock(&c.lock)
 if c.closed != 0 {
 unlock(&c.lock)
 panic(plainError("close of closed channel"))
 }
 if raceenabled {
 callerpc := getcallerpc()
 racewritepc(unsafe.Pointer(c), callerpc, funcPC(closechan))
 racerelease(unsafe.Pointer(c))
 }
 
 // 设置标志
 c.closed = 1
 var glist *g
 // 处理所有的接收者,注意即使关闭了,也是可以接收的,因为有缓冲,缓冲里面还有东西
 // release all readers
 for {
 sg := c.recvq.dequeue()
 if sg == nil {
 break
 }
 if sg.elem != nil {
 typedmemclr(c.elemtype, sg.elem)
 sg.elem = nil
 }
 if sg.releasetime != 0 {
 sg.releasetime = cputicks()
 }
 gp := sg.g
 gp.param = nil
 if raceenabled {
 raceacquireg(gp, unsafe.Pointer(c))
 }
 gp.schedlink.set(glist)
 glist = gp
 }
 // 但是对于发送的来说,如果你关闭了,还有人在发,那么就会无情的panic了,这个在发送的代码里面可以看到,在这里是处理所有发送的goroutine就可以了
 // release all writers (they will panic)
 for {
 sg := c.sendq.dequeue()
 if sg == nil {
 break
 }
 sg.elem = nil
 if sg.releasetime != 0 {
 sg.releasetime = cputicks()
 }
 gp := sg.g
 gp.param = nil
 if raceenabled {
 raceacquireg(gp, unsafe.Pointer(c))
 }
 gp.schedlink.set(glist)
 glist = gp
 }
 unlock(&c.lock)
 // Ready all Gs now that we've dropped the channel lock.
 for glist != nil {
 gp := glist
 glist = glist.schedlink.ptr()
 gp.schedlink = 0
 goready(gp, 3)
 }
}

在这里总结一下出现 panic 的情况:

1. close 一个 nil 的 channel
2. close 一个已经 closed 的 channel
3. 向一个 closed 的 channel 发送消息

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传