golang channel执行原理与代码分析

使用的go版本为 go1.21.2

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

package main

import "fmt"

func main() {
    ch := make(chan struct{})

    go func() {
        ch <- struct{}{}
        ch <- struct{}{}
    }()

    fmt.Println("xiaochuan", <-ch)

    data, ok := <-ch
    fmt.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 *sudog
    last  *sudog
}

type sudog struct {
    // 下面的字段由 sudog 阻塞的 channel 的 hchan.lock 保护。
    // shrinkstack 依赖这个字段来处理参与 channel 操作的 sudog。

    g *g

    next *sudog
    prev *sudog
    elem unsafe.Pointer // 数据元素（可能指向堆栈）

    // 下面的字段在任何情况下都不会并发访问。
    // 对于 channels，waitlink 只有 g 访问。
    // 对于 semaphores，所有字段（包括上面的字段）
    // 仅在持有 semaRoot 锁时才会访问。

    acquiretime int64
    releasetime int64
    ticket      uint32

    // isSelect 表示 g 参与了 select，因此 g.selectDone 必须进行 CAS 操作以赢得唤醒竞争。
    isSelect bool

    // success 表示通信是否成功。如果 goroutine 被唤醒是因为在通道 c 上传递了值，则为 true，
    // 如果是因为 c 被关闭而唤醒，则为 false。
    success bool

    parent   *sudog // semaRoot 二叉树
    waitlink *sudog // g.waiting 列表或 semaRoot
    waittail *sudog // semaRoot
    c        *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"))
    }

    //初始化hchan
    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.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 = elem
    c.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 int64
    if 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复制给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()    //获取当前G
    //获取一个sudog, 优先从P中获取
    //如果P中的sudog缓存区（本地无锁）为空
    //从调度器层的sudog缓冲区(全局需要加锁)中拿数据放入P的sudog缓存区
    mysg := acquireSudog() 
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }

    //将sudog写入send环形队列中去
    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)

    //将当前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 = nil
    gp.activeStackChans = false
    closed := !mysg.success
    gp.param = nil //清空G的参数列表
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    mysg.c = nil
    releaseSudog(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:295

func 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函数发送数据，然后将其重置为nil
    if sg.elem != nil {
        sendDirect(c.elemtype, sg, ep)
        sg.elem = nil
    }
    //获取等待列表中的G，
    //将当前的ch解锁, sugo赋值为G当做启动参数
    gp := sg.g
    unlockf()
    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 int64
    if 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复制给ep
            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
    }

    
    gp := getg()//获取当前G
    //获取一个sudog, 优先从P中获取
    //如果P中的sudog缓存区（本地无锁）为空
    //从调度器层的sudog缓冲区(全局需要加锁)中拿数据放入P的sudog缓存区
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }

    //将sudog写入recvq环形队列中去
    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)
    //将当前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 = nil
    gp.activeStackChans = false
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    success := mysg.success
    gp.param = nil
    mysg.c = nil
    releaseSudog(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 sender
            recvDirect(c.elemtype, sg, ep)
        }
    } else {
        // 获取环形队列中的元素
        qp := chanbuf(c, c.recvx)
        if raceenabled {
            racenotify(c, c.recvx, nil)
            racenotify(c, c.recvx, sg)
        }
        // 如果数据不为空 直接ep复制给qp
        if 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 = nil
    gp := sg.g
    unlockf()
    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:358

func 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 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()
        }
        // 获取对应的G, 重置唤醒参数
        // 将这个G加入到glist中等待后续唤醒
        gp := sg.g
        gp.param = unsafe.Pointer(sg)
        sg.success = false
        if 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.g
        gp.param = unsafe.Pointer(sg)
        sg.success = false
        if raceenabled {
            raceacquireg(gp, c.raceaddr())
        }
        glist.push(gp)
    }
    unlock(&c.lock)

    // 循环glist待唤醒列表将G设置为read状态(唤醒G运行干活)
    for !glist.empty() {
        gp := glist.pop()
        gp.schedlink = 0
        goready(gp, 3)
    }
}

总结

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