前言
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 的情况：
- close 一个 nil 的 channel
- close 一个已经 closed 的 channel
- 向一个 closed 的 channel 发送消息
网友评论