chan是我们学习golang绕不开的一个话题，今天我就不讲基础的使用了，因为太多这种文章了，我讲一下channel底层的实现和它的数据结构
必须了解的数据结构
type hchan struct {
qcount uint // 所有数据
dataqsiz uint // 数据size
buf unsafe.Pointer // 指向真实数据的指针
elemsize uint16
closed uint32
elemtype *_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
}
//qcount：代表 chan 中已经接收但还没被取走的元素的个数。=len（ch）
//dataqsiz：队列的大小。chan 使用一个循环队列来存放元素
// buf：存放元素的循环队列的 buffer。
//elemtype 和 elemsize：chan 元素的类型和元素大小，一般分为普通类型和指针类型，在makechan函数中会判断是否为指针类型
//sendx：处理发送数据的指针在 buf 中的位置。一旦接收了新的数据，指针就会加上 elemsize，移向下一个位置。buf 的总大小是 elemsize 的整数倍，而且
//buf 是一个循环列表。
//recvx：处理接收请求时的指针在 buf 中的位置。一旦取出数据，此指针会移动到下一个位置。
//recvq：chan 是多生产者多消费者的模式，如果消费者因为没有数据可读而阻塞了，就会被加入到 recvq 队列中。
//sendq：如果生产者因为 buf 满了而阻塞，会被加入到 sendq 队列中。
type waitq struct {
first *sudog
last *sudog
}
// 就是gopark掉了的goroutine队列
//单个节点结构体
type sudog struct {
g *g
isSelect bool
next *sudog
prev *sudog
elem unsafe.Pointer // data element (may point to stack)
acquiretime int64
releasetime int64
ticket uint32
parent *sudog // semaRoot binary tree
waitlink *sudog // g.waiting list or semaRoot
waittail *sudog // semaRoot
c *hchan // channel
}
在说channel之前你应该了解的
用通俗一点的话来说，channel实际底层并发是靠锁lock实现的，数据写入buf和从buf读出来都有加锁的动作，
存储数据是靠一个循环队列来保存的，队列的大小就是属性buf指向的队列长度,队列长度（len）就是qcount,
实际存储数据是buf的指向的循环队列，然后sendx和recvx是来控制数据读取和删除的，顺序就是按照buf的
索引顺序来的。然后还会保存两个队列：sendq，recvq,这两个参数实际上是绑定wait状态下的g，也就是
被park掉的goroutine,要理解这句话就要有点gmp模型的概念。
1.我们创建一个buffer channel
image.png
2.此时的结构体应该是这样的
image.png
3.放入一个数据到chan
image.png
4.取走一个数据
image.png
5.那么简单的存放数据知道了，它是如何将goroutine阻塞的呢---先将一下GMP
因为这个图是网上扣下来的，不知道原图是谁的，感觉画的很不错
image.png
image.png
这里在做个简单的解释：后面专门会出一个文章来讲解一下GMP调度模型。M是一个内核线程的一个映射（实际上的原理比这个复杂许多，涉及到线程模型，后面写篇文章细讲），P是我们的调度器，一般又称为schedule，正常情况下，我们在代码执行写了一个go func,那么这个go在代码编译的时候就会被丢入一个P的队列中：就像一堆排队的地鼠G去领钱，P就是包工头，第一个地鼠把自己的挖的洞的照片给P看，P看完就给地鼠拍拍灰（准备环境），就让G去M里面去领钱，（领钱就是执行），M呢是个小金库，但是每个地鼠只有10ms时间去领钱，过了就会被p强制的给退出来，有的地鼠不用10ms就能自己退出来
chan如何将goroutine阻塞的
image.png
这里需要了解一下gopark：将goroutine进行休眠
此时GMP应该是这个样子
image.png
此时G就没有人执行了，所以就看起来像卡住了一样，那么就有个问题了：怎么让它继续运行呢，我们不是将chan读出来goroutine又能继续执行了吗？
此时Chan的结构体变成了这个样子
image.png
哪些被游离的gorouine 就被chan的队列给抓住了，放到它的发送或者等待队列中。
此时这里sendq/recvq长这个样子
image.png
这里怎么读呢：我就简单解释一下，没这个图，我想自己画一个发现太丑了
此时buf的被别人取走了一个元素，那么就从sendqpop一个出来，是从头部开始pop，然后元素放到了buf，goroutine被设置成runable状态，然后放到P的runable队列中去继续执行下文。
上面是说的发送chan被占满
如果先来读chan被阻塞，chan内部长什么样子呢
image.png
那么chan怎么处理先读后写的这种场景
按照上面的套路，其实我们可以想，写的时候，先写到buf，然后recvq检测到buf有值了，将buf 的数据pop出来，将goroutine唤醒，数据写入elem指针指向的地址。但是有没有更好的套路呢？
其实根据场景：先有等待者，然后发送者来了，就像去银行存钱一样，取钱的人来了，但是银行没钱，后面存钱的人来了。最大的不同就是gorotuine不需要到chan，就想存钱取钱的人可以不用在银行操作，那么是不是存钱的人就可以直接把钱给取钱的人
实际上的代码也是这么来处理的，就是send 的goroutine直接写了recv的stack，就减少了lock的启动和释放，提高了性能。但是也只有only operations in runtime where this happen：运行时发生这样的情况。
- 基本上所有的流程就是上面这个图片，下面是比较重要的一些方法，可以按照上面理解去下面看看，不过最好能用goland调试一下，看代码很难看懂，我在看源码的时候，确认chan 数据类型是指针还是普通数据就踩坑了，初次创建的时候调用下面的makechan，我是直接跑到runtime里面去看，发现断点没有走预期路线，我还调试了好久，才发现它的上层是反复调用这个方法，调用了3次，才返回实例，这里面我就不细细的讲了，大家有什么建议也可以评论一下，大家共同学习进步
创建chan:makechan
func makechan(t *chantype, size int) *hchan {
elem := t.elem
// compiler checks this but be safe.
if elem.size >= 1<<16 { //判断元素的大小
throw("makechan: invalid channel element type")
}
if hchanSize%maxAlign != 0 || elem.align > maxAlign {
thro("makechan: bad alignment")
}
mem, overflow := math.MulUintptr(elem.size, uintptr(size))
if overflow || mem > maxAlloc-hchanSize || size < 0 {
panic(plainError("makechan: size out of range"))
}
// 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 mem == 0:
// Queue or element size is zero.
c = (*hchan)(mallocgc(hchanSize, nil, true))
// Race detector uses this location for synchronization.
c.buf = c.raceaddr()
case elem.ptrdata == 0:
// Elements do not contain pointers.
// Allocate hchan and buf in one call.
c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
c.buf = add(unsafe.Pointer(c), hchanSize)
default:
// Elements contain pointers.
c = new(hchan)
c.buf = mallocgc(mem, elem, true)
}
c.elemsize = uint16(elem.size)
c.elemtype = elem
c.dataqsiz = uint(size)
lockInit(&c.lock, lockRankHchan)
if debugChan {
print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")
}
return c
}
chansend
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
if c == nil {
if !block {
return false
}
gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
throw("unreachable")
}
if debugChan {
print("chansend: chan=", c, "\n")
}
if raceenabled {
racereadpc(c.raceaddr(), callerpc, funcPC(chansend))
}
if !block && c.closed == 0 && full(c) {
return false
}
var t0 int64
if blockprofilerate > 0 {
t0 = cputicks()
}
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
}
// Block on the channel. Some receiver will complete our operation for us.
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)
// Signal to anyone trying to shrink our stack that we're about
// to park on a channel. The window between when this G's status
// changes and when we set gp.activeStackChans is not safe for
// stack shrinking.
atomic.Store8(&gp.parkingOnChan, 1)
gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)
// Ensure the value being sent is kept alive until the
// receiver copies it out. The sudog has a pointer to the
// stack object, but sudogs aren't considered as roots of the
// stack tracer.
KeepAlive(ep)
// someone woke us up.
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
gp.activeStackChans = false
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
}
close chan
func closechan(c *hchan) {
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(c.raceaddr(), callerpc, funcPC(closechan))
racerelease(c.raceaddr())
}
c.closed = 1
var glist gList
// 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, c.raceaddr())
}
glist.push(gp)
}
// 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, c.raceaddr())
}
glist.push(gp)
}
unlock(&c.lock)
// Ready all Gs now that we've dropped the channel lock.
for !glist.empty() {
gp := glist.pop()
gp.schedlink = 0
goready(gp, 3)
}
}
chanrecv方法
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// raceenabled: don't need to check ep, as it is always on the stack
// or is new memory allocated by reflect.
if debugChan {
print("chanrecv: chan=", c, "\n")
}
if c == nil {
if !block {
return
}
gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
throw("unreachable")
}
// Fast path: check for failed non-blocking operation without acquiring the lock.
if !block && empty(c) {
if atomic.Load(&c.closed) == 0 {
return
}
if empty(c) {
// The channel is irreversibly closed and empty.
if raceenabled {
raceacquire(c.raceaddr())
}
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
}
var t0 int64
if blockprofilerate > 0 {
t0 = cputicks()
}
lock(&c.lock)
if c.closed != 0 && c.qcount == 0 {
if raceenabled {
raceacquire(c.raceaddr())
}
unlock(&c.lock)
if ep != nil {
typedmemclr(c.elemtype, ep)
}
return true, false
}
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 {
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)
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)
}
closed := gp.param == nil
gp.param = nil
mysg.c = nil
releaseSudog(mysg)
return true, !closed
}
有疑问加站长微信联系（非本文作者）
