Channel 是golang语言自身提供的一种非常重要的语言特性， 它是实现任务执行队列、 协程间消息传递、高并发框架的基础。关于channel的用法的文章已经很多， 本文从channel源码的实现的角度， 讨论一下其实现原理。
关于channel放在： src/runtime/chan.go
channel的关键的结构体放在hchan里面， 它记录了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
}
创建channel
用法： ch := make(chan TYPE， size int)；
这里， type是channel里面传递的elem的类型；
size是channel缓存的大小：
如果为1， 代表非缓冲的channel， 表明channel里面最多有一个elem， 剩余的只能在channel外排队等待；
如果为0，
其实现代码如下：
func makechan(t *chantype, size int) *hchan {
// Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.
// buf points into the same allocation, elemtype is persistent.
// SudoG’s are referenced from their owning thread so they can’t be collected.
// TODO(dvyukov,rlh): Rethink when collector can move allocated objects.
var c *hchan
switch {
case size == 0 || elem.size == 0:
// Queue or element size is zero.
c = (*hchan)(mallocgc(hchanSize, nil, true))
// Race detector uses this location for synchronization.
c.buf = unsafe.Pointer(c)
case elem.kind&kindNoPointers != 0:
// Elements do not contain pointers.
// Allocate hchan and buf in one call.
c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// Elements contain pointers.
c = new(hchan)
c.buf = mallocgc(uintptr(size)*elem.size, elem, true)
}
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; elemalg=", elem.alg, "; dataqsiz=", size, "\n")
}
return c
}
写入channel elem
用法： ch <- elem
channel是阻塞时的管道， 从channel读取的时候，可能发生如下3种情况：
channel 已经关闭， 发生panic；
从已经收到的队列内读取一个elem；
从缓存队列内读取elem；
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic(plainError("send on closed channel"))
}
if sg := c.recvq.dequeue(); sg != nil {
// Found a waiting receiver. We pass the value we want to send
// directly to the receiver, bypassing the channel buffer (if any).
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
}
从channel elem读取elem
用法： elem, ok := <- ch
if !ok {
fmt.Println(“ch has been closed”)
}
3种可能返回值：
chan已经被关闭， OK为false， elem为该类型的空值；
如果chan此时没有值存在， 该读取语句会一直等待直到有值；
如果chan此时有值， 读取正确的值， ok为true；
代码实现：
// chanrecv receives on channel c and writes the received data to ep.
// ep may be nil, in which case received data is ignored.
// If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// 正常的channel读取流程， 直接读取
if sg := c.sendq.dequeue(); sg != nil {
// Found a waiting sender. If buffer is size 0, receive value
// directly from sender. Otherwise, receive from head of queue
// and add sender's value to the tail of the queue (both map to
// the same buffer slot because the queue is full).
recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
return true, true
}
// 从缓存的elem队列读取一个elem
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
}
// channel没有可以读取的elem， 更新channel内部状态， 等待新的elem；
}
关闭channel
用法： close(ch)
作用: 关闭channel， 并将channel内缓存的elem清除；
代码实现：
func closechan(c *hchan) {
var glist *g
// release all readers
for {
sg := c.recvq.dequeue()
gp := sg.g
gp.param = nil
gp.schedlink.set(glist)
glist = gp
}
// release all writers (they will panic)
for {
sg := c.sendq.dequeue()
sg.elem = nil
gp := sg.g
gp.param = nil
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)
}
}
每天坚持学习1小时Go语言，大家加油，我是彬哥，下期见！如果文章中不同观点、意见请文章下留言或者关注下方订阅号反馈！
社区交流群：221273219
Golang语言社区论坛 ：
www.Golang.Ltd
LollipopGo游戏服务器地址：
https://github.com/Golangltd/LollipopGo
社区视频课程课件GIT地址：
https://github.com/Golangltd/codeclass
Golang语言社区
