【Go进阶—数据结构】Channel golang

channel 是 Golang 提供的 goroutine 间的通信方式，可以让一个 goroutine 发送特定值到另一个 goroutine。
特性
通道没有缓冲区，或者有缓冲区但缓冲区没有数据时，从通道读取数据会阻塞，直到有协程向通道中写入数据。类似地，通道没有缓冲区，或者缓冲区已满时，向通道写入数据也会阻塞，直到有协程从通道读取数据。对于值为 nil 的通道，无论读写都会阻塞，而且是永久阻塞。
使用内置函数 close 可以关闭通道，尝试向已关闭的通道发送数据会触发 panic，但此时仍然可读。通道读取的表达式最多有两个返回值：
x, ok := <-ch
第一个变量表示读出的数据，第二个变量表示是否成功读取了数据，它的值只跟通道缓冲区中是否有数据有关，与通道的关闭状态无关。
实现原理
数据结构源码 src/runtime/chan.go:hchan 定义了 channel 的数据结构：

type hchan struct { qcountuint// 当前队列中剩余元素个数 dataqsiz uint// 环形队列长度，即可以存放的元素个数 bufunsafe.Pointer // 环形队列指针 elemsize uint16// 每个元素的大小 closeduint32// 标识关闭状态 elemtype *_type// 元素类型 sendxuint// 队列下标，指示元素写入时存放到队列中的位置 recvxuint// 队列下标，指示元素从队列的该位置读出 recvqwaitq// 等待读消息的 goroutine 队列 sendqwaitq// 等待写消息的 goroutine 队列 lock mutex// 互斥锁，chan 不允许并发读写 }

可以看出 channel 由队列、类型信息、goroutine 等待队列组成。
环形队列 【【Go进阶—数据结构】Channel】channel 内部实现了一个环形队列作为其缓冲区，队列的长度是创建 channel 时指定的。下图展示了一个可缓存 6 个元素的 channel 示意图：

文章图片

dataqsiz 表明了队列长度为6，即可缓存6个元素；
buf 指向队列的内存地址；
qcount 表示队列中还有两个元素；
sendx 表示后续写入的数据存储的位置，取值为 [0, 6)；
recvx 表示读取数据的位置, 取值为[0, 6)。

类型信息一个 channel 只能传递一种类型的值：

elemtype 代表类型，用于数据传递过程中的赋值；
elemsize 代表类型大小，用于在buf中定位元素位置。

等待队列从 channel 读取数据时，如果没有缓冲区或者缓冲区为空，则当前协程会被阻塞，并被加入 recvq 队列。向 channel 写入数据时，如果没有缓冲区或者缓冲区已满，则当前协程同样会被阻塞，然后加入到 sendq 的队列。处于等待队列中的协程会在其他协程操作 channel 时被唤醒。
下图展示了一个没有缓冲区的 channel，并有几个协程正在阻塞等待读取数据：

文章图片

相关操作创建通道创建 channel 的过程实际上就是初始化 hchan 结构，类型信息和缓冲区长度由 make 语句传入，buf 的大小则由元素大小和缓冲区长度共同决定。
源码 src/runtime/chan.go 中定义了创建 channel 的函数 makechan()，精简版的代码如下所示：

func makechan(t *chantype, size int) *hchan { mem, overflow := math.MulUintptr(elem.size, uintptr(size))var c *hchan switch { case mem == 0: c = (*hchan)(mallocgc(hchanSize, nil, true)) c.buf = c.raceaddr() case elem.ptrdata =https://www.it610.com/article/= 0: c = (*hchan)(mallocgc(hchanSize+mem, nil, true)) c.buf = add(unsafe.Pointer(c), hchanSize) default: c = new(hchan) c.buf = mallocgc(mem, elem, true) }c.elemsize = uint16(elem.size) c.elemtype = elem c.dataqsiz = uint(size)return c }

发送数据发送数据的操作最终都转化成了 chansend() 函数，主要代码和逻辑如下：

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool { // 如果通道为 nil，非阻塞式发送的话直接返回 false，否则将当前协程挂起 if c == nil { if !block { return false } gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2) throw("unreachable") } // 对于非阻塞式发送，如果通道未关闭且没有缓冲空间的话，直接返回 false if !block && c.closed == 0 && full(c) { return false }// 加锁，并发安全 lock(&c.lock)// 如果通道关闭了，直接 panic if c.closed != 0 { unlock(&c.lock) panic(plainError("send on closed channel")) }// 如果接收队列不为空，直接将要发送的数据发送到队首的 goroutine if sg := c.recvq.dequeue(); sg != nil { send(c, sg, ep, func() { unlock(&c.lock) }, 3) return true }// 对于缓冲区还有空闲的 channel，拷贝数据到缓冲区，维护相关信息 if c.qcount < c.dataqsiz { 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 }// 没有缓冲空间时，发送方会挂起，并根据当前 goroutine 构造一个 sudog 结构体添加到 sendq 队列中 gp := getg() 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 gp.waiting = mysg gp.param = nil c.sendq.enqueue(mysg)gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)// 省略被唤醒时部分代码return true }

读取数据读取数据的操作最终是转化成了 chanrecv() 函数，主要逻辑如下：

// selected 和 received 返回值分别代表是否可被 select 语句命中以及是否读取到了数据 func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { // 如果 channel 为 nil，非阻塞式读取直接返回，否则直接挂起 if c == nil { if !block { return } gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2) throw("unreachable") }// 非阻塞模式并且没有消息可读（没有缓冲区或者缓冲区为空），如果 channel 未关闭直接返回 if !block && empty(c) { if atomic.Load(&c.closed) == 0 { return }if empty(c) { if raceenabled { raceacquire(c.raceaddr()) } if ep != nil { typedmemclr(c.elemtype, ep) } return true, false } }// 加锁 lock(&c.lock)// channel 已关闭并且没有消息可读（没有缓冲区或者缓冲区为空），会接收到零值，typedmemclr 会根据类型清理相应地址的内存 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 }// 等待发送队列不为空，如果是非缓冲型 channel，直接拷贝发送者的数据，否则接收队首的数据，并将发送者的数据移动到环形队列尾部 if sg := c.sendq.dequeue(); sg != nil { recv(c, sg, ep, func() { unlock(&c.lock) }, 3) return true, true }// 缓冲型 channel，buf 里有元素，可以正常接收 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 }// 被阻塞的情况，构造一个 sudog 结构体，保存到 channel 的等待接收队列，并将当前 goroutine 挂起 gp := getg() mysg := acquireSudog() mysg.releasetime = 0 if t0 != 0 { mysg.releasetime = -1 }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)// 省略被唤醒时部分代码return true, !closed }

关闭通道关闭某个 channel，最终会执行函数 closechan()，核心代码如下：

func closechan(c *hchan) { // 如果 channel 为 nil，直接 panic if c == nil { panic(plainError("close of nil channel")) }// 加锁，如果 channel 已关闭，直接 panic lock(&c.lock) if c.closed != 0 { unlock(&c.lock) panic(plainError("close of closed channel")) }c.closed = 1var glist gList// 释放等待接收队列中，向需要返回值的接收者返回相应的零值 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) }// 释放等待发送队列，相关的 goroutine 会触发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)// ... }

常见应用
定时任务这种用法需要与 timer 结合，分为两种：超时控制和定时执行。
如果需要执行某项操作，但又不想它耗费太长时间，想给它一个超时限制，可以这么做：

select { case <-time.After(100 * time.Millisecond): case <-s.stopc: return false }

等待 100 ms 后，如果 s.stopc 还没有读出数据或者被关闭，就直接结束。
定时执行某个任务也比较简单，例如每隔 1 秒种，执行一次定时任务：

func worker() { ticker := time.Tick(1 * time.Second) for { select { case <- ticker: // 执行任务 } } }

解耦生产者与消费者使用一个 channel 保存任务，启动 n 个 goroutine 作为工作协程池，这些协程工作在一个无限循环里，从该 channel 读取任务并执行：

func main() { taskCh := make(chan int, 100) go worker(taskCh)for i := 0; i < 10; i++ { taskCh <- i }select { case <-time.After(time.Hour): } } func worker(taskCh <-chan int) { const N = 5for i := 0; i < N; i++ { go func(id int) { for { task := <- taskCh fmt.Printf("finish task: %d by worker %d\n", task, id) time.Sleep(time.Second) } }(i) } }

控制并发数有时需要定时执行几百个任务，但是并发数又不能太高，这时就可以通过 channel 来控制并发数。比如下面的例子：

var limit = make(chan int, 3)func main() { // ………… for _, w := range work { go func() { limit <- 1 w() <-limit }() } // ………… }

构建一个容量为 3 的 channel，遍历任务列表，每个任务启动一个 goroutine，真正执行任务的动作在 w() 中完成。在执行 w() 之前，先要从 limit 中拿“许可证”，拿到许可证之后，才能执行 w()，并且在执行完任务，要将“许可证”归还。要注意的是，如果 w() 发生 panic，那“许可证”可能就还不回去了，因此需要使用 defer 来保证。