🧑‍🏫 Go 中同步组件 Chan 的理解.

深入Chan底层源码进行分析Chan工作原理.

学习比较枯燥，但贵在坚持.  有关于源码的理解我都已经写在源码中的注释中了。Go源码看起来要比Java舒服多了，仅仅只是对于环境上，哈哈哈，Java 看个Spring源码麻烦的一批.

🏷️ 一、向Chan中发送数据.

1.1 总览全局：chan 发送数据的几个步骤.

• 第一种情况：chan关闭状态，直接panic，表示不能向已经关闭状态的chan读取数据.

• 第二种情况：直接查看当前chan中等待接受数据的等待队列是否为空，如果不为空直接发送.

• 第三种情况：查看当前与chan相关的环形队列是否为空，如果不为空则将当前要进行发送的数据放入进去.

• 第四种情况：如果是非阻塞状态直接进行return，如果是阻塞状态，那么就将当前的goroutine添加到chan关联的等待发送数据的等待队列中.

1.2 源码分析.

不管是通过x<-1进行发送还是通过select关键字发送，其最后调用runtime.chansend1，而该方法内部就是对函数chansend的一次调用. 所以说chan发送数据最终就是通过chansend函数来进行发送.

func chansend1(c *hchan, elem unsafe.Pointer) {
    chansend(c, elem, true, getcallerpc())
}

下面我们先看看整体的源码.

点我查看主要流程图

下面就是我们的源码分析, 这里我会把无关的源码和注释清理掉

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {

    // chan发送数据时一定会调用的方法.
    if c == nil {
        if !block {
            return false
        }
        gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }
    if !block && c.closed == 0 && full(c) {
        // 非阻塞.
        return false
    }

    lock(&c.lock)
    // 0应该是未关闭状态.
    if c.closed != 0 {
        unlock(&c.lock)
        panic(plainError("send on closed channel"))
    }

    // 第一步: 不管环形队列中是否有空闲的位置，我现在进行发送，如果有等待接收的goroutine,那么我直接将当前要进行发送的数据发送到等待队列中的G.
    if sg := c.recvq.dequeue(); sg != nil {
        send(c, sg, ep, func() { unlock(&c.lock) }, 3)
        return true
    }

    // 第二步: 当不存在等待接收的G时,并且环形队列中存在剩余的空间,这个时候直接写入Chan的缓冲区.
    if c.qcount < c.dataqsiz {
        // Space is available in the channel buffer. Enqueue the element to send.
        // 计算可以进行存储的下标.
        qp := chanbuf(c, c.sendx)
        if raceenabled {
            racenotify(c, c.sendx, nil)
        }
        // 将发送的数据拷贝到缓冲区.
        // ep 就是本次要进行发送的数据.
        typedmemmove(c.elemtype, qp, ep)
        // 移动下标.
        c.sendx++
        if c.sendx == c.dataqsiz {
            c.sendx = 0
        }
        // 增加count.
        c.qcount++
        unlock(&c.lock)
        return true
    }

    // 是否为阻塞.
    // 使用select关键字可以向Chan非阻塞的发送消息.
    // select 关键字触发的chan写block为false.
    // 阻塞的话,会让出CPU等待时间.
    if !block {
        // select 关键字会走这个.
        // 非阻塞的话,CPU不会让出执行时间.
        unlock(&c.lock)
        return false
    }

    // 等待队列中既没有可以接受的goroutine.
    // 环形队列又没有可以存储数据的剩余空间,这个时候将当前的goroutine添加到等待发送队列.
    // 然后进行gopack.
    // 下面实现起来就很复杂.
    // Block on the channel. Some receiver will complete our operation for us.
    // getg获取当前发送数据使用的Goroutine.
    gp := getg()
    // 获取表示当前G的结构体：runtime.sudog.并设置这一次阻塞发送的相关消息.
    mysg := acquireSudog()
    mysg.releasetime = 0
    if t0 != 0 {
        mysg.releasetime = -1
    }
    mysg.elem = ep
    mysg.waitlink = nil
    mysg.g = gp
    mysg.isSelect = false
    mysg.c = c
    // 将刚刚创建的sudog加入发送等待队列，并设置当前的Goroutine的waiting上.
    gp.waiting = mysg
    gp.param = nil
    // 将表示当前的Goroutine加入到当前Chan中的等待发送队列中.
    c.sendq.enqueue(mysg)
    atomic.Store8(&gp.parkingOnChan, 1)
    // 将当前的goroutine陷入沉睡并且等待唤醒.
    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)

    // 表示当前的goroutine被唤醒之后汇之星一些收尾工作,将一些属性置为零值，并且释放runtime.sudog结构体.
    // sudog 结构体应该和goroutine是有关联的.
    // someone woke us up.
    if mysg != gp.waiting {
        throw("G waiting list is corrupted")
    }
    // 下面就是无关代码.
    gp.waiting = nil
    gp.activeStackChans = false
    closed := !mysg.success
    gp.param = nil
    if mysg.releasetime > 0 {
        blockevent(mysg.releasetime-t0, 2)
    }
    mysg.c = nil
    releaseSudog(mysg)
    if closed {
        if c.closed == 0 {
            throw("chansend: spurious wakeup")
        }
        panic(plainError("send on closed channel"))
    }
    return true
}

🏷️ 二、从Chan中读取数据.

2.1总览全局：从chan中接受数据

• 第一种情况：非阻塞，缓存为空，直接return.

• 第二种情况：chan关闭，且缓存为空 return.

• 第三种情况：如果chan上，待发送的队列队头不为nil，直接获取该队头上的数据.

• 第四种情况：如果环形队列qcount不为0，则直接从环形队列上进行获取.

• 第五种情况：如果是非阻塞，那么直接return 如果是阻塞状态，那么将会进入等待接受数据的阻塞等待队列.

2.2 源码分析.

Go 中通过chan发送数据，最终都会调用chanrecv函数来完成发送. 下面的代码我会删掉无用代码以及无用注释.

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    // chan 接受数据.
    // chan结构体为nil和chan结构体关闭两种情况.
    if c == nil {
        // block 为true 表示是阻塞状态.
        if !block {
            // 非阻塞状态直接return,这里直接返回默认零值.
            return
        }
        // gopark表示睡眠.
        gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }

    // block为true表示阻塞.
    // empty判断的是chan中的环形队列中的容器.
    // Fast path: check for failed non-blocking operation without acquiring the lock.
    // 如果是非阻塞,且缓存为空,那么直接return.
    if !block && empty(c) {
        // 非阻塞且为空  那么剩余的就应该是: 非阻塞且不为空.   阻塞的其他状态.
        // CAS 加载chan是否为关闭状态.
        if atomic.Load(&c.closed) == 0 {
            // 非阻塞环形队列为空,且未关闭,直接return false.
            return
        }
        // 如果chan关闭.
        if empty(c) {
            // chan中的环形队列容量为0 或者就是容量不为0,但是qcount为0.
            // The channel is irreversibly closed and empty.
            if raceenabled {
                raceacquire(c.raceaddr())
            }
            // 发送的数据不为nil.
            if ep != nil {
                typedmemclr(c.elemtype, ep)
            }
            // 有个问题：这里也没有判断发送队列是不是为空.
            // Go中应该只有select才会触发非阻塞.
            return true, false
        }
    }

    // 代码能运行到这里就说要么是非阻塞且缓存不为空要么就是阻塞状态.
    var t0 int64
    if blockprofilerate > 0 {
        t0 = cputicks()
    }

    lock(&c.lock)

    // closed 为 0表示未关闭.
    // chan关闭,但是缓存为空.
    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 {
            racenotify(c, c.recvx, nil)
        }
        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
    }

    // 如果是非阻塞状态, 直接return.
    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
    }

    // 将要接受数据的goroutine添加到要接受数据的等待队列中.
    // 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)
    // 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), 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)
    }
    success := mysg.success
    gp.param = nil
    mysg.c = nil
    releaseSudog(mysg)
    return true, success
}