go学习---Chan（通道）Go语言如何判断一个chan被关闭

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

Golang使用Groutine和channels实现了CSP(Communicating Sequential Processes)模型，channles在goroutine的通信和同步中承担着重要的角色。在GopherCon 2017中，Golang专家Kavya深入介绍了 Go Channels 的内部机制，以及运行时调度器和内存管理系统是如何支持Channel的

以一个简单的channel应用开始，使用goroutine和channel实现一个任务队列，并行处理多个任务。

func main(){
    //带缓冲的channel
    ch := make(chan Task, 3)

    //启动固定数量的worker
    for i := 0; i< numWorkers; i++ {
        go worker(ch)
    }

    //发送任务给worker
    hellaTasks := getTaks()

    for _, task := range hellaTasks {
        ch <- task
    }

    ...
}

func worker(ch chan Task){
    for {
       //接受任务
       task := <- ch
       process(task)
    }
}

从上面的代码可以看出，使用golang的goroutine和channel可以很容易的实现一个生产者-消费者模式的任务队列，相比java, c++简洁了很多。channel可以天然的实现了下面四个特性：
- goroutine安全
- 在不同的goroutine之间存储和传输值
- 提供FIFO语义(buffered channel提供)
- 可以让goroutine block/unblock

那么channel是怎么实现这些特性的呢？下面我们看看当我们调用make来生成一个channel的时候都做了些什么。

make chan

上述任务队列的例子第三行，使用make创建了一个长度为3的带缓冲的channel，channel在底层是一个hchan结构体，位于src/runtime/chan.go里。其定义如下:

type hchan struct {
    qcount   uint           // total data in the queue
    dataqsiz uint           // size of the circular queue
    buf      unsafe.Pointer // points to an array of dataqsiz elements
    elemsize uint16
    closed   uint32
    elemtype *_type // element type
    sendx    uint   // send index
    recvx    uint   // receive index
    recvq    waitq  // list of recv waiters
    sendq    waitq  // list of send waiters

    // 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
}

make函数在创建channel的时候会在该进程的heap区申请一块内存，创建一个hchan结构体，返回执行该内存的指针，所以获取的的ch变量本身就是一个指针，在函数之间传递的时候是同一个channel。

hchan结构体使用一个环形队列来保存groutine之间传递的数据(如果是缓存channel的话)，使用两个list保存像该chan发送和从改chan接收数据的goroutine，还有一个mutex来保证操作这些结构的安全。

向channel发送和从channel接收数据主要涉及hchan里的四个成员变量，借用Kavya ppt里的图示，来分析发送和接收的过程。

send

　发送到channel上面去，还有buf则放入。没有则放入到对应的sendq等待队列中。

recv

　从buf中取数据。没有则放入到对应的recvq等待队列中。

关于阻塞队列

关于放入阻塞队列到底是如何实现的。需要了解到GO的调度。Go语言调度有3个比较重要的概念。对于OS来说，一般是进程调度，线程调度。里面有调度器，调度算法等来实现，现在比较经典的是公平调度算法，详细见 https://blog.csdn.net/u012279631/article/details/77677266。go语言相当于把这个权利自己拿过来了。在用户态自己调用。在线程之下挂载了协程称之为G(goroutine),而线程称之为M(machine 即底层)，具体调度者称之为P(context)。这是因为进程切换上下文耗费的系统资源大，创建线程的时候耗费8K内存。当在高并发的情况下去处理对应的事务，如果用线程去处理仍然对资源非常浪费。从这个角度来讲，golang就是为了处理高并发而特定的一种语言。

线程和协程映射关系

　P从runable队列中取到相应的G，放到M中取执行。

阻塞情形

channel队列满无法放入阻塞

当channel已经满了，仍然有数据发送到channel中时。

原本状态，G1在running中。

发现无法放入channel中，导致阻塞。

这时候会从runable队列中取下一个goroutine，放到M中去执行。

再回头看channel相应的改变。因为buffer已经满了。所以会把它放在sendq中。

结构如下：

当有goroutine去取对应buffer时，清空一个buffer。就会把sendx里面的elem内容copy到对应的channel buffer中。然后把goroutine状态设置为run，并且放回到runable队列中。

channel队列空无法读取阻塞

这里用了非常聪明的方式减少了一次内存的拷贝。

当G1发送内容到channel的时候，首先查看recvq队列是否有阻塞的goroutine。如果有则直接从G1copy到G2。优化了从G1 -> channel ->G2这个步骤。

参考链接：

https://speakerdeck.com/kavya719/understanding-channels

https://tiancaiamao.gitbooks.io/go-internals/content/zh/07.1.html
---------------------
还是以前面的任务队列为例:

//G1
func main(){
    ...

    for _, task := range hellaTasks {
        ch <- task    //sender
    }

    ...
}

//G2
func worker(ch chan Task){
    for {
       //接受任务
       task := <- ch  //recevier
       process(task)
    }
}

其中G1是发送者，G2是接收，因为ch是长度为3的带缓冲channel，初始的时候hchan结构体的buf为空，sendx和recvx都为0，当G1向ch里发送数据的时候，会首先对buf加锁，然后将要发送的数据copy到buf里，并增加sendx的值，最后释放buf的锁。然后G2消费的时候首先对buf加锁，然后将buf里的数据copy到task变量对应的内存里，增加recvx，最后释放锁。整个过程，G1和G2没有共享的内存，底层通过hchan结构体的buf，使用copy内存的方式进行通信，最后达到了共享内存的目的，这完全符合CSP的设计理念

一般情况下，G2的消费速度应该是慢于G1的，所以buf的数据会越来越多，这个时候G1再向ch里发送数据，这个时候G1就会阻塞，那么阻塞到底是发生了什么呢？

Goroutine Pause/Resume

goroutine是Golang实现的用户空间的轻量级的线程，有runtime调度器调度，与操作系统的thread有多对一的关系.如前面图所示，

M是操作系统的线程，G是用户启动的goroutine，P是与调度相关的context，每个M都拥有一个P，P维护了一个能够运行的goutine队列，用于该线程执行。

当G1向buf已经满了的ch发送数据的时候，当runtine检测到对应的hchan的buf已经满了，会通知调度器，调度器会将G1的状态设置为waiting, 移除与线程M的联系，然后从P的runqueue中选择一个goroutine在线程M中执行，此时G1就是阻塞状态，但是不是操作系统的线程阻塞，所以这个时候只用消耗少量的资源。

那么G1设置为waiting状态后去哪了？怎们去resume呢？我们再回到hchan结构体，注意到hchan有个sendq的成员，其类型是waitq，查看源码如下：

Go 
type hchan struct { 
... 
recvq waitq // list of recv waiters 
sendq waitq // list of send waiters 
... 
} 
// 
type waitq struct { 
first *sudog 
last *sudog 
}

实际上，当G1变为waiting状态后，会创建一个代表自己的sudog的结构，然后放到sendq这个list中，sudog结构中保存了channel相关的变量的指针(如果该Goroutine是sender，那么保存的是待发送数据的变量的地址，如果是receiver则为接收数据的变量的地址，之所以是地址，前面我们提到在传输数据的时候使用的是copy的方式
当G2从ch中接收一个数据时，会通知调度器，设置G1的状态为runnable，然后将加入P的runqueue里，等待线程执行.

前面我们是假设G1先运行，如果G2先运行会怎么样呢？如果G2先运行，那么G2会从一个empty的channel里取数据，这个时候G2就会阻塞，和前面介绍的G1阻塞一样，G2也会创建一个sudog结构体，保存接收数据的变量的地址，但是该sudog结构体是放到了recvq列表里，当G1向ch发送数据的时候，runtime并没有对hchan结构体题的buf进行加锁，而是直接将G1里的发送到ch的数据copy到了G2 sudog里对应的elem指向的内存地址！

chan的分类

分为带缓存和不带缓存这2类，尤其需要关注带缓存的用法，防止掉坑里。

不带缓存：make(chan 数据类型)
带缓存： make(chan 数据类型,长度)

例如定义一个带缓存的chan：

ch := make(chan int,2)

这里我们定义个缓存长度为2的chan，当我们已经往chan中写入了2个数据，当再次写入第三个数据的时候就会发送阻塞，直到其他人从该chan中读取了数据，那么才可以再次写入数据，带缓存的chan类似于一个队列，当队列满的时候是无法写入数据的。

chan的关闭

chan可以通过close关闭，关闭后的chan是无法写入数据的，但是可以读取数据。

当一个chanel被关闭后，再取出不会阻塞，而是返回零值

package main
 
import "fmt"
 
func main() {
    c := make(chan int, 5)
    c <- 123
    close(c)
     
    fmt.Println(<-c)
    fmt.Println(<-c)

输出

123
0

判断的方法是否关闭方法就是接收第二个参数，如下

package main
 
import "fmt"
 
func main() {
    c := make(chan int, 10)
    c <- 123
    close(c)
 
    var res int
    var ok bool
 
    res, ok = <-c
    fmt.Println(res, ok)
 
    res, ok = <-c
    fmt.Println(res, ok) //此时ok为false
}

　　输出：

123 true
0 false

问题：无缓存chan,在一个goroutine向chan塞了数据之后，然后当前goroutine就是堵塞，必须由另外一个goutine来取走chan数据就会接着走下面的流程。为什么必须要另外一个goroutine才能取走数据呢，为啥不能自己塞自己取，比如在main goroutine 向chan 塞数据后，然后立马再去拿出来，就会报错产生死锁。

是因为当chan堵塞的时候，会把当前导致堵塞的goroutine 置为一个waiting的状态，也就说当前的这个goroutine自己已经不可能再接受chan的数据了，必须是由另外的一个goroutine 接收之后，才能继续走下面的流程代码。