golang channel执行原理与代码分析

使用的go版本为 go1.21.2

首先我们写一个简单的chan调度代码

package mainimport "fmt"func main() {ch := make(chan struct{})go func() {ch <- struct{}{}ch <- struct{}{}}()fmt.Println("xiaochuan", <-ch)data, ok := <-chfmt.Println("xiaochuan", data, ok)close(ch)
}

因为ch的数据获取方式有两种，所以这个示例代码写了两次的ch读与写
老样子通过go build -gcflags -S main.go获取到对应的汇编代码

调度make最终被转换为CALL runtime.makechan

调度ch <- struct{}{}最终被转换为CALL runtime.chansend1 由于我们调度了两次所以这里有两个 

 调度 <-ch 最终被转换为CALL runtime.chanrecv1

我们还进行一次两个参数的调度接收ch读取
data, ok := <-ch最终被转换为CALL runtime.chanrecv2 

 调度 close(ch) 最终被转换为CALL runtime.closechan 先来看一下hchan构造体相关的底层源码

hchan结构体

//代码位于 GOROOT/src/runtime/chan.go L:33
type hchan struct {qcount   uint           // 环形队列中元素个数dataqsiz uint           // 环形队列的大小buf      unsafe.Pointer // 指向大小为 dataqsiz 的数组elemsize uint16         // 元素大小closed   uint32         // 是否关闭elemtype *_type         // 元素类型sendx    uint           // 发送索引recvx    uint           // 接收索引recvq    waitq          // recv 等待列表，即（ <-ch ）sendq    waitq          // send 等待列表，即（ ch<- ）lock     mutex          // 锁
}type waitq struct { // 等待队列 sudog 双向队列first *sudoglast  *sudog
}type sudog struct {// 下面的字段由 sudog 阻塞的 channel 的 hchan.lock 保护。// shrinkstack 依赖这个字段来处理参与 channel 操作的 sudog。g *gnext *sudogprev *sudogelem unsafe.Pointer // 数据元素（可能指向堆栈）// 下面的字段在任何情况下都不会并发访问。// 对于 channels，waitlink 只有 g 访问。// 对于 semaphores，所有字段（包括上面的字段）// 仅在持有 semaRoot 锁时才会访问。acquiretime int64releasetime int64ticket      uint32// isSelect 表示 g 参与了 select，因此 g.selectDone 必须进行 CAS 操作以赢得唤醒竞争。isSelect bool// success 表示通信是否成功。如果 goroutine 被唤醒是因为在通道 c 上传递了值，则为 true，// 如果是因为 c 被关闭而唤醒，则为 false。success boolparent   *sudog // semaRoot 二叉树waitlink *sudog // g.waiting 列表或 semaRootwaittail *sudog // semaRootc        *hchan // channel
}

先从创建chan开始

makechan源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:65//如果我们make的初始化缓冲区比较大会调度这个函数
func makechan64(t *chantype, size int64) *hchan {//将size强转为int类型//因为go的int类型的大小在不同平台上可能是 32 位或 64 位//如果大小超过了当前平台int最大值，会截断掉超出最大值的部分if int64(int(size)) != size {panic(plainError("makechan: size out of range"))}//强制转换为int类型超出int部分截断return makechan(t, int(size))
}func makechan(t *chantype, size int) *hchan {elem := t.Elem//编辑器检测元素的大小会不会大于2的16次方，对齐方式if elem.Size_ >= 1<<16 {throw("makechan: invalid channel element type")}if hchanSize%maxAlign != 0 || elem.Align_ > maxAlign {throw("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"))}//初始化hchanvar c *hchanswitch {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.PtrBytes == 0: //元素不包含指针(在调用中分配 hchan 和 buf)// 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 = elemc.dataqsiz = uint(size)lockInit(&c.lock, lockRankHchan)if debugChan { //开启debug开关，公屏打印print("makechan: chan=", c, "; elemsize=", elem.Size_, "; dataqsiz=", size, "\n")}return c
}

chansend1源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:142
//c <- x 调度这个函数
func chansend1(c *hchan, elem unsafe.Pointer) {chansend(c, elem, true, getcallerpc())
}func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {if c == nil { //判断当前ch是不是一个空指针，如果为空将当前G休眠，触发崩溃if !block {return false}gopark(nil, nil, waitReasonChanSendNilChan, traceBlockForever, 2)throw("unreachable")}if debugChan { //开启debug开关，公屏打印print("chansend: chan=", c, "\n")}if raceenabled {//竞争开启racereadpc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(chansend))}//在无锁的情况下，检测一下是否ch 是否关闭，是否会造成阻塞if !block && c.closed == 0 && full(c) {return false}var t0 int64if blockprofilerate > 0 {t0 = cputicks()}lock(&c.lock) //获取chan锁if c.closed != 0 { // 二次确认chan是不是已经关闭unlock(&c.lock)panic(plainError("send on closed channel"))}//判断当前ch是否存在接收方//如果存在直接调用send函数将数据发送给对方，避免数据复制到缓存区中去if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3)return true}//判断当前ch元素个数是否小于队列的长度//如果有剩余空间将数据将要发送的元素加入队列if c.qcount < c.dataqsiz {// 获取环形队列中的元素qp := chanbuf(c, c.sendx)if raceenabled {racenotify(c, c.sendx, nil)}// 直接ep复制给qptypedmemmove(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()    //获取当前G//获取一个sudog, 优先从P中获取//如果P中的sudog缓存区（本地无锁）为空//从调度器层的sudog缓冲区(全局需要加锁)中拿数据放入P的sudog缓存区mysg := acquireSudog() mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}//将sudog写入send环形队列中去mysg.elem = epmysg.waitlink = nilmysg.g = gpmysg.isSelect = falsemysg.c = cgp.waiting = mysggp.param = nilc.sendq.enqueue(mysg)//将当前G的parkingOnChan设置为true（表示目前停止在了chansend或chanrecv上）//将当前的G移出调度队列（调度chanparkcommit解锁当前ch）gp.parkingOnChan.Store(true)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceBlockChanSend, 2)//调度KeepAlive函数确保发送的元素处于一个可达的状态避免被回收KeepAlive(ep)//当前后续唤醒G//判断G的等待列表是否为当前的sudog//如果不一致说明G已经被改写了if mysg != gp.waiting {throw("G waiting list is corrupted")}//清空G的等待队列，//获取当前被唤醒的原因sudog.succes//因为唤醒方式有两种，1。通道关闭 2.接收唤起gp.waiting = nilgp.activeStackChans = falseclosed := !mysg.successgp.param = nil //清空G的参数列表if mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}mysg.c = nilreleaseSudog(mysg) //释放sudog重新放回P的sudogcache(本地)if closed { //由于不能写入关闭的chan，所以直接异常了if c.closed == 0 {throw("chansend: spurious wakeup")}panic(plainError("send on closed channel"))}return true
}

直接发送的时候调用的send函数解读如下

send源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:295func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {if raceenabled {if c.dataqsiz == 0 {racesync(c, sg)} else {// Pretend we go through the buffer, even though// we copy directly. Note that we need to increment// the head/tail locations only when raceenabled.racenotify(c, c.recvx, nil)racenotify(c, c.recvx, sg)c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz}}// 检测数据是否为空// 如果不为空直接调用sendDirect函数发送数据，然后将其重置为nilif sg.elem != nil {sendDirect(c.elemtype, sg, ep)sg.elem = nil}//获取等待列表中的G，//将当前的ch解锁, sugo赋值为G当做启动参数gp := sg.gunlockf()gp.param = unsafe.Pointer(sg)sg.success = true//sugo判断释放时间是否为0//为0将其设置为当前 CPU 的时钟滴答数if sg.releasetime != 0 {sg.releasetime = cputicks()}//将G标记为可运行状态,放入调度队列等待被后续调度goready(gp, skip+1)
}

chanrecv1与chanrecv2源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:442//chanrecv1与chanrecv2的处理逻辑基本差不多
//chanrecv2多接受了一个变量而已 
//可以理解为这样ok := chanrecv2(ch, v)
func chanrecv1(c *hchan, elem unsafe.Pointer) {chanrecv(c, elem, true)
}func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {_, received = chanrecv(c, elem, true)return
}func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {if debugChan {//开启debug开关，公屏打印print("chanrecv: chan=", c, "\n")}if c == nil {//判断当前ch是不是为空指针，如果为空将当前G休眠，触发崩溃if !block {return}gopark(nil, nil, waitReasonChanReceiveNilChan, traceBlockForever, 2)throw("unreachable")}if !block && empty(c) {//非阻塞情况下, 且数据队列为空if atomic.Load(&c.closed) == 0 { //原子读取 当前ch是否关闭，如果关闭直接返回// Because a channel cannot be reopened, the later observation of the channel// being not closed implies that it was also not closed at the moment of the// first observation. We behave as if we observed the channel at that moment// and report that the receive cannot proceed.return}if empty(c) {// 重新检测是否为空ch// The channel is irreversibly closed and empty.if raceenabled {raceacquire(c.raceaddr())}if ep != nil {typedmemclr(c.elemtype, ep)}return true, false}}var t0 int64if blockprofilerate > 0 {t0 = cputicks()}lock(&c.lock) //获取chan锁if c.closed != 0 {  // 二次确认ch是不是已经关闭if c.qcount == 0 {if raceenabled {raceacquire(c.raceaddr())}unlock(&c.lock)if ep != nil {typedmemclr(c.elemtype, ep)}return true, false}} else {// 判断当前ch是否存在发送方// 如果存在直接调用recv函数将数据接受对方的数据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}}//环形队列中存在数据，直接从队列中接收，传递给接受者if c.qcount > 0 {// 获取环形队列中的元素qp := chanbuf(c, c.recvx)if raceenabled {racenotify(c, c.recvx, nil)}if ep != nil {// 直接qp复制给eptypedmemmove(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}gp := getg()//获取当前G//获取一个sudog, 优先从P中获取//如果P中的sudog缓存区（本地无锁）为空//从调度器层的sudog缓冲区(全局需要加锁)中拿数据放入P的sudog缓存区mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}//将sudog写入recvq环形队列中去mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.isSelect = falsemysg.c = cgp.param = nilc.recvq.enqueue(mysg)//将当前G的parkingOnChan设置为true（表示目前停止在了chansend或chanrecv上）//将当前的G移出调度队列（调度chanparkcommit解锁当前ch）gp.parkingOnChan.Store(true)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceBlockChanRecv, 2)//当前后续唤醒G//判断G的等待列表是否为当前的sudog//如果不一致说明G已经被改写了if mysg != gp.waiting {throw("G waiting list is corrupted")}//清空G的等待队列，//获取当前被唤醒的原因sudog.succes//因为唤醒方式有两种，1。通道关闭 2.发送唤起gp.waiting = nilgp.activeStackChans = falseif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}success := mysg.successgp.param = nilmysg.c = nilreleaseSudog(mysg)//释放sudog重新放回P的sudogcache(本地)return true, success
}

直接读取的时候调用的recv函数解读如下

recv源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:616
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {//判断当前环形队列是否为0//为0从发送方复制数据（调度recvDirect函数）if c.dataqsiz == 0 { if raceenabled {racesync(c, sg)}if ep != nil {// copy data from senderrecvDirect(c.elemtype, sg, ep)}} else {// 获取环形队列中的元素qp := chanbuf(c, c.recvx)if raceenabled {racenotify(c, c.recvx, nil)racenotify(c, c.recvx, sg)}// 如果数据不为空 直接ep复制给qpif ep != nil {typedmemmove(c.elemtype, ep, qp)}// 清除数据typedmemmove(c.elemtype, qp, sg.elem)c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz}//获取等待列表中的G，//将当前的ch解锁, sugo赋值为G当做启动参数sg.elem = nilgp := sg.gunlockf()gp.param = unsafe.Pointer(sg)sg.success = true//sugo判断释放时间是否为0//为0将其设置为当前 CPU 的时钟滴答数if sg.releasetime != 0 {sg.releasetime = cputicks()}//将G标记为可运行状态,放入调度队列等待被后续调度goready(gp, skip+1)
}

closechan源码与解读

//代码位于 GOROOT/src/runtime/chan.go L:358func closechan(c *hchan) {if c == nil {//如果ch未初始化直接报错panic(plainError("close of nil channel"))}lock(&c.lock) //获取chan锁if c.closed != 0 { //如果当前ch已经处于关闭状态,触发异常unlock(&c.lock)panic(plainError("close of closed channel"))}if raceenabled { //竞争开启callerpc := getcallerpc()racewritepc(c.raceaddr(), callerpc, abi.FuncPCABIInternal(closechan))racerelease(c.raceaddr())}c.closed = 1 //将当前ch设置为关闭状态//待唤醒的G列表var glist gList// release all readersfor { //逐步从读取队列取值,直到获取完为止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()}// 获取对应的G, 重置唤醒参数// 将这个G加入到glist中等待后续唤醒gp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseif raceenabled {raceacquireg(gp, c.raceaddr())}glist.push(gp)}for {//逐步从发送队列取值,直到获取完为止 （向关闭的ch发送数据会有panic）sg := c.sendq.dequeue()if sg == nil {break}sg.elem = nil// 重置释放时间if sg.releasetime != 0 {sg.releasetime = cputicks()}// 获取对应的G, 重置唤醒参数// 将这个G加入到glist中等待后续唤醒gp := sg.ggp.param = unsafe.Pointer(sg)sg.success = falseif raceenabled {raceacquireg(gp, c.raceaddr())}glist.push(gp)}unlock(&c.lock)// 循环glist待唤醒列表将G设置为read状态(唤醒G运行干活)for !glist.empty() {gp := glist.pop()gp.schedlink = 0goready(gp, 3)}
}

总结

我们从上面的源码分析了解chan的数据结构、发送数据、接收数据和关闭这些基本操作，从源码分析我们得知chan的读写操作是会上锁的，如果业务中对性能要求比较高的情况下chan的这把锁会成为我们系统内的瓶颈。