Channel

设计原理

不要通过共享内存的方式进行通信，而是应该通过通信的方式共享内存。

在主流编程语言中，多个线程传递数据的方式一般都是共享内存。

Go 可以使用共享内存加互斥锁进行通信，同时也提供了一种不同的并发模型，即通信顺序进程（Communicating sequential processes，CSP）。Goroutine 和 Channel 分别对应 CSP 中的实体和传递信息的媒介，Goroutine 之间会通过 Channel 传递数据。
上图中的两个 Goroutine，一个会向 Channel 中发送数据，另一个会从 Channel 中接收数据，它们两者能够独立运行并不存在直接关联，但是能通过 Channel 间接完成通信。

接收数据

两个 Goroutine，一个会向 Channel 中发送数据，另一个会从 Channel 中接收数据，它们两者能够独立运行并不存在直接关联，但是能通过 Channel 间接完成通信。这是一个 生产者 - 消费者 模型，负责接收数据的 goroutine 从 channel 读取一个消息进行消费，channel 起到一个临界区/缓冲区的作用。

// chanrecv 函数接收 channel c 的元素并将其写入 ep 所指向的内存地址。
// 如果 ep 是 nil，说明忽略了接收值。
// 如果 block == false，即非阻塞型接收，在没有数据可接收的情况下，返回 (false, false)
// 否则，如果 c 处于关闭状态，将 ep 指向的地址清零，返回 (true, false)
// 否则，用返回值填充 ep 指向的内存地址。返回 (true, true)
// 如果 ep 非空，则应该指向堆或者函数调用者的栈func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {// 省略 debug 内容 …………// 如果是一个 nil 的 channelif c == nil {// 如果不阻塞，直接返回 (false, false)if !block {return}// 否则，接收一个 nil 的 channel，goroutine 挂起gopark(nil, nil, "chan receive (nil chan)", traceEvGoStop, 2)// 不会执行到这里throw("unreachable")}// 在非阻塞模式下，快速检测到失败，不用获取锁，快速返回// 当我们观察到 channel 没准备好接收：// 1. 非缓冲型，等待发送列队 sendq 里没有 goroutine 在等待// 2. 缓冲型，但 buf 里没有元素// 之后，又观察到 closed == 0，即 channel 未关闭。// 因为 channel 不可能被重复打开，所以前一个观测的时候 channel 也是未关闭的，// 因此在这种情况下可以直接宣布接收失败，返回 (false, false)if !block && (c.dataqsiz == 0 && c.sendq.first == nil ||c.dataqsiz > 0 && atomic.Loaduint(&c.qcount) == 0) &&atomic.Load(&c.closed) == 0 {return}var t0 int64if blockprofilerate > 0 {t0 = cputicks()}// 加锁lock(&c.lock)// channel 已关闭，并且循环数组 buf 里没有元素// 这里可以处理非缓冲型关闭 和 缓冲型关闭但 buf 无元素的情况// 也就是说即使是关闭状态，但在缓冲型的 channel，// buf 里有元素的情况下还能接收到元素if c.closed != 0 && c.qcount == 0 {if raceenabled {raceacquire(unsafe.Pointer(c))}// 解锁unlock(&c.lock)if ep != nil {// 从一个已关闭的 channel 执行接收操作，且未忽略返回值// 那么接收的值将是一个该类型的零值// typedmemclr 根据类型清理相应地址的内存typedmemclr(c.elemtype, ep)}// 从一个已关闭的 channel 接收，selected 会返回truereturn true, false}// 等待发送队列里有 goroutine 存在，说明 buf 是满的// 1. 非缓冲型的 channel。直接进行内存拷贝（从 sender goroutine -> receiver goroutine）// 2. 缓冲型的 channel，但 buf 满了。接收到循环数组头部的元素，并将发送者的元素放到循环数组尾部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}// 缓冲型，buf 里有元素，可以正常接收if c.qcount > 0 {// 直接从循环数组里找到要接收的元素qp := chanbuf(c, c.recvx)// …………// 没有忽略要接收的值，不是 "<- ch"，而是 "val <- ch"，ep 指向 valif ep != nil {typedmemmove(c.elemtype, ep, qp)}// 清理掉循环数组里相应位置的值typedmemclr(c.elemtype, qp)// 接收游标向前移动c.recvx++// 接收游标归零if c.recvx == c.dataqsiz {c.recvx = 0}// buf 数组里的元素个数减 1c.qcount--// 解锁unlock(&c.lock)return true, true}if !block {// 非阻塞接收，解锁。selected 返回 false，因为没有接收到值unlock(&c.lock)return false, false}// 构造一个 sudog 并设置相应参数gp := getg()mysg := acquireSudog()mysg.releasetime = 0if t0 != 0 {mysg.releasetime = -1}mysg.elem = epmysg.waitlink = nilgp.waiting = mysgmysg.g = gpmysg.selectdone = nilmysg.c = cgp.param = nil// 进入 channel 的等待接收队列c.recvq.enqueue(mysg)// 将当前 goroutine 挂起goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)// 被唤醒了，接着从这里继续执行一些扫尾工作if mysg != gp.waiting {throw("G waiting list is corrupted")}gp.waiting = nilif mysg.releasetime > 0 {blockevent(mysg.releasetime-t0, 2)}closed := gp.param == nilgp.param = nilmysg.c = nil// 释放当前 gorountine 的 sudogreleaseSudog(mysg)return true, !closed
}func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {// 如果是非缓冲型的 channelif c.dataqsiz == 0 {if raceenabled {racesync(c, sg)}// 未忽略接收的数据 不是 "<- ch"，而是 "val <- ch"，ep 指向 valif ep != nil {// 直接拷贝数据，从 sender goroutine -> receiver goroutinerecvDirect(c.elemtype, sg, ep)}} else {// 缓冲型的 channel，但 buf 已满。// 1. 循环数组 buf 队首的元素拷贝到接收数据的地址// 2. 将 sender 的数据入队。qp := chanbuf(c, c.recvx)// …………// 将 recvx 处的数据拷贝给接收者if ep != nil {typedmemmove(c.elemtype, ep, qp)}// sender data -> buftypedmemmove(c.elemtype, qp, sg.elem)// 更新索引c.recvx++if c.recvx == c.dataqsiz {c.recvx = 0}c.sendx = c.recvx}sg.elem = nilgp := sg.g// 解锁unlockf()gp.param = unsafe.Pointer(sg)if sg.releasetime != 0 {sg.releasetime = cputicks()}// 将当前处理器的 runnext 设置成发送数据的 Goroutine，在调度器下一次调度时将阻塞的发送方唤醒。goready(gp, skip+1)
}func recvDirect(t *_type, sg *sudog, dst unsafe.Pointer) {// dst is on our stack or the heap, src is on another stack.src := sg.elemtypeBitsBulkBarrier(t, uintptr(dst), uintptr(src), t.size)memmove(dst, src, t.size)
}

从 channel 接收消息 的 核心函数是 chanrecv

跟 send 流程差不多：
特殊情况：

1. 如果 channel 为空，那么会直接调用 runtime.gopark 挂起当前 goroutine。
2. 如果 channel 已经关闭并且缓冲区没有任何数据，runtime.chanrecv 会直接返回零值。
   正常情况：
3. 如果 channel 的 sendq 队列中存在挂起的 goroutine，会将 recvx 索引所在的数据拷贝到接收变量所在的内存空间上并将 sendq 队列中 goroutine 的数据拷贝到缓冲区。
4. 如果 channel 的缓冲区中包含数据，那么直接读取 recvx 索引对应的数据。
5. 在默认情况下会挂起当前的 goroutine，将 runtime.sudog 结构加入 recvq 队列并陷入休眠等待调度器的唤醒。
   从 channel 接收数据时，会触发 goroutine 调度的两个时机：
6. 当 channel 为空时。
7. 当缓冲区中不存在数据并且也不存在数据的发送者时。