深入理解go语言channel

69 阅读 0 评论 46 点赞

我是靠谱客的博主英俊万宝路，最近开发中收集的这篇文章主要介绍深入理解go语言channel，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

channel 是 Go 语言中的一个非常重要的特性，这篇文章来深入了解一下 channel。

1. CSP

要想理解 channel 要先知道 CSP 模型。CSP 是 Communicating Sequential Process 的简称，中文可以叫做通信顺序进程，是一种并发编程模型，由 Tony Hoare 于 1977 年提出。简单来说，CSP 模型由并发执行的实体（线程或者进程）所组成，实体之间通过发送消息进行通信，这里发送消息时使用的就是通道，或者叫 channel。CSP 模型的关键是关注 channel，而不关注发送消息的实体。Go 语言实现了 CSP 部分理论，goroutine 对应 CSP 中并发执行的实体，channel 也就对应着 CSP 中的 channel。

2. channel 基础知识

2.1 创建 channel

channel 使用之前需要通过 make 创建。

unBufferChan := make(chan int)
// 1
bufferChan := make(chan int, N) // 2

上面的方式 1 创建的是无缓冲 channel，方式 2 创建的是缓冲 channel。如果使用 channel 之前没有 make，会出现 dead lock 错误。

func main() {
var x chan intgo func() {
x <- 1
}()
<-x
}

$ go run channel1.go
fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan receive (nil chan)]:
main.main()
/Users/kltao/code/go/examples/channl/channel1.go:11 +0x60
goroutine 4 [chan send (nil chan)]:
main.main.func1(0x0)

2.2 channel 读写操作

ch := make(chan int, 10)
// 读操作
x <- ch
// 写操作
ch <- x

2.3 channel 种类

channel 分为无缓冲 channel 和有缓冲 channel。两者的区别如下：

无缓冲：发送和接收动作是同时发生的。如果没有 goroutine 读取 channel （<- channel），则发送者 (channel <-) 会一直阻塞。
缓冲：缓冲 channel 类似一个有容量的队列。当队列满的时候发送者会阻塞；当队列空的时候接收者会阻塞。

2.4 关闭 channel

channel 可以通过 built-in 函数 close() 来关闭。

ch := make(chan int)
// 关闭
close(ch)

关于关闭 channel 有几点需要注意的是：

重复关闭 channel 会导致 panic。
向关闭的 channel 发送数据会 panic。
从关闭的 channel 读数据不会 panic，读出 channel 中已有的数据之后再读就是 channel 类似的默认值，比如 chan int 类型的 channel 关闭之后读取到的值为 0。

对于上面的第三点，我们需要区分一下：channel 中的值是默认值还是 channel 关闭了。可以使用 ok-idiom 方式，这种方式在 map 中比较常用。

ch := make(chan int, 10)
...
close(ch)
// ok-idiom
val, ok := <-ch
if ok == false {
// channel closed
}

3. channel 的典型用法

3.1 goroutine 通信

func main() {
x := make(chan int)
go func() {
x <- 1
}()
<-x
}

3.2 select

select 一定程度上可以类比于 linux 中的 IO 多路复用中的 select。后者相当于提供了对多个 IO 事件的统一管理，而 Golang 中的 select 相当于提供了对多个 channel 的统一管理。当然这只是 select 在 channel 上的一种使用方法。

select {
case e, ok := <-ch1:
...
case e, ok := <-ch2:
...
default:
}

值得注意的是，当select 用于多个channel监听并收发消息，当任何一个case满足条件则会执行，如果有多个channel队列上都有数据流，则随机选取一个channel；若没有可执行的case，就会执行default，如果没有default，程序就会阻塞。

3.3 range channel

range channel 可以直接取到 channel 中的值。当我们使用 range 来操作 channel 的时候，一旦 channel 关闭，channel 内部数据读完之后循环自动结束。

func consumer(ch chan int) {
for x := range ch {
fmt.Println(x)
...
}
}
func producer(ch chan int) {
for _, v := range values {
ch <- v
}
}

3.4. 单向 channel

单向 channel，顾名思义只能写或读的 channel。但是仔细一想，只能写的 channel，如果不读其中的值有什么用呢？其实单向 channel 主要用在函数声明中。比如。

func foo(ch chan<- int) <-chan int {...}

foo 的形参是一个只能写的 channel，那么就表示函数 foo 只会对 ch 进行写，当然你传入的参数可以是个普通 channel。foo 的返回值是一个只能读的 channel，那么表示 foo 的返回值规范用法就是只能读取。这种写法在 Golang 的原生代码库中有非常多的示例，感兴趣的可以去看一下。

4. channel 源码分析

如果忽略内置的 channel，让你设计一个具有 goroutines-safe 并且可以用来存储、传递值的东西，你会怎么做？很多人可能觉得或许可以用一个带锁的队列来做。没错，事实上，channel 内部就是一个带锁的队列。

type hchan struct {
qcount
uint
// 队列中数据个数
dataqsiz uint
// channel 大小
buf
unsafe.Pointer // 存放数据的环形数组
elemsize uint16
// channel 中数据类型的大小
closed
uint32
// 表示 channel 是否关闭
elemtype *_type // 元素数据类型
sendx
uint
// send 的数组索引
recvx
uint
// recv 的数组索引
recvq
waitq
// 由 recv 行为（也就是 <-ch）阻塞在 channel 上的 goroutine 队列
sendq
waitq
// 由 send 行为 (也就是 ch<-) 阻塞在 channel 上的 goroutine 队列
// lock protects all fields in hchan, as well as several// fields in sudogs blocked on this channel. Do not change another G's status while holding this lock// (in particular, do not ready a G), as this can deadlock// with stack shrinking.
lock mutex
}

buf 的具体实现很简单，就是一个环形队列的实现。sendx 和 recvx 分别用来记录发送、接收的位置。然后用一个 lock 互斥锁来确保无竞争冒险。对于每一个 ch := make(chan Task, 3) 这类操作，都会在堆中，分配一个空间，建立并初始化一个 hchan 结构变量，而 ch 则是指向这个 hchan 结构的指针。因为 ch 本身就是个指针，所以我们才可以在 goroutine 函数调用的时候直接将 ch 传递过去，而不用再 &ch 取指针了，所以所有使用同一个 ch 的 goroutine 都指向了同一个实际的内存空间。

4.1 channel对goroutine 的发送与接收

为了方便描述，我们用 G1 表示 main() 函数的 goroutine，而 G2 表示 worker 的 goroutine。

// G1
func main() {
...
for _, task := range tasks {
ch <- task
}
...
}
// G2
func worker(ch chan Task) {
for {
task :=<-ch
process(task)
}
}

那么 G1 中的 ch <- task0 具体是怎么做的呢？

获取锁
enqueue(task0)（这里是内存复制 task0）
释放锁

这一步很简单，接下来看 G2 的 t := <- ch 是如何读取数据的。

获取锁
t = dequeue()（同样，这里也是内存复制）
释放锁

这一步也非常简单。但是我们从这个操作中可以看到，所有 goroutine 中共享的部分只有这个 hchan 的结构体，而所有通讯的数据都是内存复制。这遵循了 Go 并发设计中很核心的一个理念：Do not communicate by sharing memory;instead, share memory by communicating

发送数据分三种情况：

有 goroutine 阻塞在 channel 上，此时 hchan.buf 为空：直接将数据发送给该 goroutine。
当前 hchan.buf 还有可用空间：将数据放到 buffer 里面。
当前 hchan.buf 已满：阻塞当前 goroutine。

4.2 channel对goroutine 的阻塞和恢复

发送方被阻塞：

还是用刚刚G1和G2的例子，假设 G2 需要很长时间的处理，在此期间，G1 不断的发送任务：

ch <- task1
ch <- task2
ch <- task3

但是当再一次 ch <- task4 的时候，由于 ch 的缓冲只有 3 个，所以没有地方放了，于是 G1 被 block 了，当有人从队列中取走一个 Task 的时候，G1 才会被恢复。这是我们都知道的，不过我们今天关心的不是发生了什么，而是如何做到的？

goroutine 被阻塞的具体过程 ：

那么当 ch <- task4 执行的时候，channel 中已经满了，需要pause G1。这个时候，：G1 会调用运行的 gopark，然后 Go 的运行时调度器就会接管，将 G1 的状态设置为 waiting，断开 G1 和 M 之间的关系（switch out)，因此 G1 脱离 M，换句话说，M 空闲了，可以安排别的任务了。从 P 的运行队列中，取得一个可运行的 goroutine G

建立新的 G 和 M 的关系（Switch in)，因此 G 就准备好运行了。当调度器返回的时候，新的 G 就开始运行了，而 G1 则不会运行，也就是 block 了。从上面的流程中可以看到，对于 goroutine 来说，G1 被阻塞了，新的 G 开始运行了；而对于操作系统线程 M 来说，则根本没有被阻塞。我们知道 OS 线程要比 goroutine 要沉重的多，因此这里尽量避免 OS 线程阻塞，可以提高性能。

goroutine 恢复执行的具体过程 ：

前面理解了阻塞，那么接下来理解一下如何恢复运行。不过，在继续了解如何恢复之前，我们需要先进一步理解 hchan 这个结构。因为，当 channel 不在满的时候，调度器是如何知道该让哪个 goroutine 继续运行呢？而且 goroutine 又是如何知道该从哪取数据呢？

在 hchan 中，除了之前提到的内容外，还定义有 sendq 和 recvq 两个队列，分别表示等待发送、接收的 goroutine，及其相关信息。

type waitq struct {
first *sudog
last
*sudog
}
type sudog struct {
// The following fields are protected by the hchan.lock of the
// channel this sudog is blocking on. shrinkstack depends on
// this for sudogs involved in channel ops.
g
*g
selectdone *uint32 // CAS to 1 to win select race (may point to stack)
next
*sudog
prev
*sudog
elem
unsafe.Pointer // data element (may point to stack)
// The following fields are never accessed concurrently.
// For channels, waitlink is only accessed by g.
// For semaphores, all fields (including the ones above)
// are only accessed when holding a semaRoot lock.
acquiretime int64
releasetime int64
ticket
uint32
parent
*sudog // semaRoot binary tree
waitlink
*sudog // g.waiting list or semaRoot
waittail
*sudog // semaRoot
c
*hchan // channel
}

所以在之前的阻塞 G1 的过程中，实际上：

G1 会给自己创建一个 sudog 的变量
然后追加到 sendq 的等候队列中，方便将来的 receiver 来使用这些信息恢复 G1。

这些都是发生在调用调度器之前，那么现在开始看一下如何恢复：

当 G2 调用 t := <- ch 的时候，channel 的状态是，缓冲是满的，而且还有一个 G1 在等候发送队列里，然后 G2 执行下面的操作：

G2 先执行 dequeue() 从缓冲队列中取得 task1 给 t
G2 从 sendq 中弹出一个等候发送的 sudog
将弹出的 sudog 中的 elem 的值 enqueue() 到 buf 中
将弹出的 sudog 中的 goroutine，也就是 G1，状态从 waiting 改为 runnable
然后，G2 需要通知调度器 G1 已经可以进行调度了，因此调用 goready(G1)。
调度器将 G1 的状态改为 runnable，调度器将 G1 压入 P 的运行队列，因此在将来的某个时刻调度的时候，G1 就会开始恢复运行。返回到 G2

注意，这里是由 G2 来负责将 G1 的 elem 压入 buf 的，这是一个优化。这样将来 G1 恢复运行后，就不必再次获取锁、enqueue()、释放锁了。这样就避免了多次锁的开销。

如果接收方先阻塞:

如果 G2 先执行了 t := <- ch，此时 buf 是空的，因此 G2 会被阻塞，他的流程是这样：

1.G2 给自己创建一个 sudog 结构变量。其中 g 是自己，也就是 G2，而 elem 则指向 t

2.将这个 sudog 变量压入 recvq 等候接收队列

3.G2 需要告诉 goroutine，自己需要 pause 了，于是调用 gopark(G2)

和之前一样，调度器将其 G2 的状态改为 waiting
断开 G2 和 M 的关系
从 P 的运行队列中取出一个 goroutine
建立新的 goroutine 和 M 的关系
返回，开始继续运行新的 goroutine

这些应该已经不陌生了，那么当 G1 开始发送数据的时候，流程是什么样子的呢？

G1 可以将 enqueue(task)，然后调用 goready(G2)。不过，我们可以更聪明一些。我们根据 hchan 结构的状态，已经知道 task 进入 buf 后，G2 恢复运行后，会读取其值，复制到 t 中。那么 G1 可以根本不走 buf，G1 可以直接把数据给 G2。Goroutine 通常都有自己的栈，互相之间不会访问对方的栈内数据，除了 channel。这里，由于我们已经知道了 t 的地址（通过 elem指针），而且由于 G2 不在运行，所以我们可以很安全的直接赋值。当 G2 恢复运行的时候，既不需要再次获取锁，也不需要对 buf 进行操作。从而节约了内存复制、以及锁操作的开销。