Go 学习笔记（25）— 并发（04）[有缓冲/无缓冲通道、WaitGroup 协程同步、select 多路监听通道、close 关闭通道、channel 传参或作为结构体成员]

1. 无缓冲的通道

无缓冲的通道（unbuffered channel）是指在接收前没有能力保存任何值的通道。

这种类型的通道要求发送 goroutine 和接收 goroutine 同时准备好，才能完成发送和接收操作。

如果两个 goroutine 没有同时准备好，通道会导致先执行发送或接收操作的 goroutine 阻塞等待。这种对通道进行发送和接收的交互行为本身就是同步的。其中任意一个操作都无法离开另一个操作单独存在。

下图展示两个 goroutine 如何利用无缓冲的通道来共享一个值。

两个 goroutine 都到达通道，但两者都没有开始执行发送或者接收。
左侧的 goroutine 将它的手伸进了通道，这模拟了向通道发送数据的行为。这时，这个 goroutine 会在通道中被锁住，直到交换完成。
右侧的 goroutine 将它的手放入通道，这模拟了从通道里接收数据。这个 goroutine 一样也会在通道中被锁住，直到交换完成。
进行交换。
右侧的 goroutine 拿到数据。
两个 goroutine 都将它们的手从通道里拿出来，这模拟了被锁住的 goroutine 得到释放。两个 goroutine 现在都可以去做别的事情了。

图：使用无缓冲的通道在 goroutine 之间同步，摘自《Go 语言实战》

package mainimport ("runtime"
)func main() {c := make(chan struct{})go func(i chan struct{}) {sum := 0for i := 0; i <= 10000; i++ {sum += i}println("sum is :", sum)// 写通道c <- struct{}{}}(c)//NumGoroutine 可以返回当前程序的 goroutine 数目println("NumGoroutine=", runtime.NumGoroutine())// 读取通道 c, 通过通道进行同步等待<-c
}

无缓冲通道需要发送和接收配对。否则会被阻塞，直到另一方准备好后被唤醒。

package mainimport "fmt"func main() {data := make(chan int)  // 数据交换队列exit := make(chan bool) // 退出通知go func() {for d := range data { // 从队列迭代接收数据，直到 close 。fmt.Println(d)}fmt.Println("recv over.")exit <- true // 发出退出通知。}()data <- 1 // 发送数据。data <- 2data <- 3close(data) // 关闭队列。fmt.Println("send over.")<-exit // 等待退出通知。
}

输出：

1
2
3
send over.
recv over.

2. 有缓冲的通道

在无缓冲通道的基础上，为通道增加一个有限大小的存储空间形成带缓冲通道。带缓冲通道在发送时无需等待接收方接收即可完成发送过程，并且不会发生阻塞，只有当存储空间满时才会发生阻塞。同理，如果缓冲通道中有数据，接收时将不会发生阻塞，直到通道中没有数据可读时，通道将会再度阻塞。

有缓冲的通道（buffered channel）是一种在被接收前能存储一个或者多个值的通道。

这种类型的通道并不强制要求 goroutine 之间必须同时完成发送和接收。通道会阻塞发送和接收动作的条件也会不同。

只有在通道中没有要接收的值时，接收动作才会阻塞。只有在通道没有可用缓冲区容纳被发送的值时，发送动作才会阻塞。

这导致有缓冲的通道和无缓冲的通道之间的一个很大的不同：

无缓冲的通道保证进行发送和接收的 goroutine 会在同一时间进行数据交换；
有缓冲的通道没有这种保证。

在下图中可以看到两个 goroutine 分别向有缓冲的通道里增加一个值和从有缓冲的通道里移除一个值。

右侧的 goroutine 正在从通道接收一个值。
右侧的 goroutine 独立完成了接收值的动作，而左侧的 goroutine 正在发送一个新值到通道里。
左侧的 goroutine 还在向通道发送新值，而右侧的 goroutine 正在从通道接收另外一个值。这个步骤里的两个操作既不是同步的，也不会互相阻塞。
所有的发送和接收都完成，而通道里还有几个值，也有一些空间可以存更多的值。

图：使用有缓冲的通道在 goroutine 之间同步数据，摘自《Go 语言实战》

有缓冲通道例子

package mainimport ("runtime"
)func main() {c := make(chan struct{})ci := make(chan int, 100)go func(i chan struct{}, j chan int) {for i := 0; i <= 10; i++ {ci <- i}close(ci)// 写通道c <- struct{}{}}(c, ci)//NumGoroutine 可以返回当前程序的 goroutine 数目println("NumGoroutine=", runtime.NumGoroutine())// 读取通道 c, 通过通道进行同步等待<-c// 此时ci 通道已经关闭，匿名函数启动的goroutine 已经退出println("NumGoroutine=", runtime.NumGoroutine())// 但是通道 ci 还可以继续读取for v := range ci {println("v is :", v)}
}

异步方式也就是有缓冲的通道通过判断缓冲区来决定是否阻塞。

缓冲区已满，发送被阻塞；
缓冲区为空，接收被阻塞；

通常情况下，异步 channel 可减少排队阻塞，具备更高的效率。但应该考虑使用指针规避大对象拷贝，将多个元素打包，减小缓冲区大小等。

为什么Go语言对通道要限制长度而不提供无限长度的通道？

我们知道通道（ channel ）是在两个 goroutine 间通信的桥梁。使用 goroutine 的代码必然有一方提供数据，一方消费数据。当提供数据一方的数据供给速度大于消费方的数据处理速度时，如果通道不限制长度，那么内存将不断膨胀直到应用崩溃。

因此，限制通道的长度有利于约束数据提供方的供给速度，供给数据量必须在消费方处理量+通道长度的范围内，才能正常地处理数据。

package mainimport "fmt"func main() {data := make(chan int, 3) // 缓冲区可以存储 3 个元素exit := make(chan bool)data <- 1 // 在缓冲区未满前，不会阻塞。data <- 2data <- 3go func() {for d := range data { // 在缓冲区未空前，不会阻塞。fmt.Println(d)}exit <- true}()data <- 4 // 如果缓冲区已满，阻塞。data <- 5close(data)<-exit
}

缓冲区是内部属性，并非类型构成要素。

var a, b chan int = make(chan int), make(chan int, 3)

除用 range 外，还可用 ok-idiom 模式判断 channel 是否关闭。

for {if d, ok := <-data; ok {fmt.Println(d)} else {break}
}

向 closed channel 发送数据引发 panic 错误，接收立即返回零值。而 nil channel，无论收发都会被阻塞。

// 这个示例程序展示如何使用
// 有缓冲的通道和固定数目的
// goroutine来处理一堆工作
package mainimport ("fmt""math/rand""sync""time"
)const (numberGoroutines = 4  // 要使用的goroutine的数量taskLoad         = 10 // 要处理的工作的数量
)// wg用来等待程序完成
var wg sync.WaitGroup// init初始化包，Go语言运行时会在其他代码执行之前
// 优先执行这个函数
func init() {// 初始化随机数种子rand.Seed(time.Now().Unix())
}// main是所有Go程序的入口
func main() {// 创建一个有缓冲的通道来管理工作tasks := make(chan string, taskLoad)// 启动goroutine来处理工作wg.Add(numberGoroutines)for gr := 1; gr <= numberGoroutines; gr++ {go worker(tasks, gr)}// 增加一组要完成的工作for post := 1; post <= taskLoad; post++ {tasks <- fmt.Sprintf("Task : %d", post)}// 当所有工作都处理完时关闭通道// 以便所有goroutine退出close(tasks)// 等待所有工作完成wg.Wait()
}// worker作为goroutine启动来处理
// 从有缓冲的通道传入的工作
func worker(tasks chan string, worker int) {// 通知函数已经返回defer wg.Done()for {// 等待分配工作task, ok := <-tasksif !ok {// 这意味着通道已经空了，并且已被关闭fmt.Printf("Worker: %d : Shutting Down\n", worker)return}// 显示我们开始工作了fmt.Printf("Worker: %d : Started %s\n", worker, task)// 随机等一段时间来模拟工作sleep := rand.Int63n(100)time.Sleep(time.Duration(sleep) * time.Millisecond)// 显示我们完成了工作fmt.Printf("Worker: %d : Completed %s\n", worker, task)}
}

输出:

Worker: 4 : Started Task : 2
Worker: 1 : Started Task : 1
Worker: 2 : Started Task : 3
Worker: 3 : Started Task : 4
Worker: 4 : Completed Task : 2
Worker: 4 : Started Task : 5
Worker: 2 : Completed Task : 3
Worker: 2 : Started Task : 6
Worker: 3 : Completed Task : 4
Worker: 3 : Started Task : 7
Worker: 3 : Completed Task : 7
Worker: 3 : Started Task : 8
Worker: 4 : Completed Task : 5
Worker: 4 : Started Task : 9
Worker: 1 : Completed Task : 1
Worker: 1 : Started Task : 10
Worker: 3 : Completed Task : 8
Worker: 3 : Shutting Down
Worker: 2 : Completed Task : 6
Worker: 2 : Shutting Down
Worker: 1 : Completed Task : 10
Worker: 1 : Shutting Down
Worker: 4 : Completed Task : 9
Worker: 4 : Shutting Down

在main函数的第31行，创建了一个string类型的有缓冲的通道，缓冲的容量是10。在第34行，给WaitGroup赋值为4，代表创建了4个工作 goroutine。之后在第35行到第37行，创建了4个 goroutine，并传入用来接收工作的通道。在第40行到第42行，将10个字符串发送到通道，模拟发给 goroutine 的工作。一旦最后一个字符串发送到通道，通道就会在第46行关闭，而main函数就会在第49行等待所有工作的完成。

第46行中关闭通道的代码非常重要。当通道关闭后，goroutine 依旧可以从通道接收数据，但是不能再向通道里发送数据。能够从已经关闭的通道接收数据这一点非常重要，因为这允许通道关闭后依旧能取出其中缓冲的全部值，而不会有数据丢失。从一个已经关闭且没有数据的通道里获取数据，总会立刻返回，并返回一个通道类型的零值。如果在获取通道时还加入了可选的标志，就能得到通道的状态信息。

在worker函数里，可以在第58行看到一个无限的for循环。在这个循环里，会处理所有接收到的工作。每个 goroutine 都会在第60行阻塞，等待从通道里接收新的工作。一旦接收到返回，就会检查ok标志，看通道是否已经清空而且关闭。如果ok的值是false，goroutine 就会终止，并调用第56行通过defer声明的Done函数，通知main有工作结束。

如果ok标志是true，表示接收到的值是有效的。第71行和第72行模拟了处理的工作。一旦工作完成，goroutine 会再次阻塞在第60行从通道获取数据的语句。一旦通道被关闭，这个从通道获取数据的语句会立刻返回，goroutine 也会终止自己。

3. WaitGroup

Go 语言中除了可以使用通道（channel）和互斥锁进行两个并发程序间的同步外，还可以使用等待组进行多个任务的同步，等待组可以保证在并发环境中完成指定数量的任务。sync.WaitGroup 类型（以下简称WaitGroup类型）是开箱即用的，也是并发安全的。

一般情况下，我会用这个方法来记录需要等待的 goroutine 的数量。相对应的，这个类型的 Done 方法，用于对其所属值中计数器的值进行减一操作。我们可以在需要等待的 goroutine 中，通过 defer 语句调用它。而此类型的 Wait 方法的功能是，阻塞当前的 goroutine ，直到其所属值中的计数器归零。如果在该方法被调用的时候，那个计数器的值就是 0，那么它将不会做任何事情。

goroutine 和 chan ，一个用于并发，另一个用于通信。没有缓冲的通道具有同步的功能，除此之外， sync 包也提供了多个 goroutine 同步的机制，主要是通过 WaitGroup 实现的。

WaitGroup 值中计数器的值不能小于 0，是因为这样会引发一个 panic 。

如果在一个此类值的 Wait 方法被执行期间，跨越了两个计数周期，那么就会引发一个 panic 。纵观上述会引发 panic 的后两种情况，我们可以总结出这样一条关于 WaitGroup 值的使用禁忌，

即：不要把增加其计数器值的操作和调用其Wait方法的代码，放在不同的 goroutine 中执行。换句话说，要杜绝对同一个WaitGroup 值的两种操作的并发执行。

我们最好用 先统一 Add ，再并发 Done ，最后 Wait 这种标准方式，来使用 WaitGroup 值。尤其不要在调用 Wait 方法的同时，并发地通过调用 Add 方法去增加其计数器的值，因为这也有可能引发 panic 。

在 sync.WaitGroup （等待组）类型中，每个 sync.WaitGroup 值在内部维护着一个计数，此计数的初始默认值为零。

主要的接口如下：

type WaitGroup struct {// contains filtered or unexported fields
}// 添加等待信号
func (wg *WaitGroup) Add(delta int)// 释放等待信号
func (wg *WaitGroup) Done()// 等待
func (wg *WaitGroup) Wait()

WaitGroup 用来等待多个 goroutine 完成；
main goroutine 调用 Add 设置需要等待 goroutine 的数目；
每一个 goroutine 结束时调用 Done() ；
Wait() 被 main 用来等待所有的 goroutine 完成；

sync.WaitGroup 内部拥有一个计数器，计数器的值可以通过方法调用实现计数器的增加和减少。当我们添加了 N 个并发任务进行工作时，就将等待组的计数器值增加 N。每个任务完成时，这个值减 1。同时，在另外一个 goroutine 中等待这个等待组的计数器值为 0 时，表示所有任务已经完成。

代码示例：

package mainimport ("net/http""sync"
)var wg sync.WaitGroup
var urls = []string{"http://www.baidu.com","http://www.sina.com","http://www.qq.com",
}func main() {for _, url := range urls {// 为每一个 url 启动一个 goroutine，同时给 wg 加 1wg.Add(1)go func(url string) {// 当前go routine 结束后给wg 计数减1, wg.Done() 等价于wg.Add(-1)// defer wg.Add(-1)defer wg.Done()// 发送 http get 请求并打印 http 返回码resp, err := http.Get(url)if err == nil {println(resp.Status)}}(url)}// 等待所有请求结束wg.Wait()
}

或者不使用匿名函数，如下

package mainimport ("net/http""sync"
)var wg sync.WaitGroup
var urls = []string{"http://www.baidu.com","http://www.sina.com","http://www.qq.com",
}func getURLStatus(url string) {// 当前go routine 结束后给wg 计数减1, wg.Done() 等价于wg.Add(-1)// defer wg.Add(-1)defer wg.Done()// 发送 http get 请求并打印 http 返回码resp, err := http.Get(url)if err == nil {println(resp.Status)}
}func main() {for _, url := range urls {// 为每一个 url 启动一个 goroutine，同时给 wg 加 1wg.Add(1)go getURLStatus(url)}// 等待所有请求结束wg.Wait()
}

4. select

select 是类 UNIX 系统提供的一个多路复用系统API， Go 语言借用多路复用的概念，提供了 select 关键字，用于多路监昕多个通道。

select 语句只能与通道联用，它一般由若干个分支组成。每次执行这种语句的时候，一般只有一个分支中的代码会被运行。

当监听的通道没有状态是可读或可写的， select 是阻塞的；只要监听的通道中有一个状态是可读或可写的，则 select 就不会阻塞，而是进入处理就绪通道的分支流程。如果监听的通道有多个可读或可写的状态，则 select 随机选取一个处理。

select 的特点是只要其中有一个 case 已经完成，程序就会继续往下执行，而不会考虑其他 case 的情况。

select 的用法与 switch 语言非常类似，由 select 开始一个新的选择块，每个选择条件由 case 语句来描述。与 switch 语句相比， select 有比较多的限制，其中最大的一条限制就是每个 case 语句里必须是一个 IO 操作。结构如下：

select{case 操作1:响应操作1case 操作2:响应操作2…default:没有操作情况
}

操作1、操作2：包含通道收发语句，请参考下表。

操作	语句示例
接收任意数据	case <- ch;
接收变量	case d := <- ch;
发送数据	case ch <- 100;

在 Go 中，支持通信操作的类型只有 chan ，所以 select 中的 case 条件只能是对 chan 类型变量的读写操作。由于 chan 类型变量的读写操作可能会引起阻塞，为了在使用 select 选择器时不陷入阻塞状态，可以在 select 代码块中添加 default 关键字，当 case 条件全部都不满足时，默认进入 default 分支，执行完 default 分支的代码后，退出 select 选择器。

package mainimport ("fmt""time"
)func main() {fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))select {case <-time.After(time.Second * 2):fmt.Println("2秒后的时间：", time.Now().Format("2006-01-02 15:04:05"))}
}

输出结果：

开始时间： 2021-02-08 14-14-42
2秒后的时间： 2021-02-08 14:14:44

time.After 函数返回一个通道类型的变量，然后在 case 中从这个通道中读取信息，如果没有协程给这个通道发送信息，那么 case 将会一直阻塞。在调用 After 函数时，传入了一个时长作为参数，意思是从调用 After 函数算起，到设定的时长后，有协程将会向这个通道发送一条消息。当通道收到消息后，这个 case 条件满足，这个 case 分支下的代码将会被执

如果没有任意一条 select 语句可以执行（即所有的通道都被阻塞），那么有如下两种可能的情况：

如果给出了 default 语句，那么就会执行 default 语句，同时程序的执行会从 select 语句后的语句中恢复；
如果没有 default 语句，那么 select 语句将被阻塞，直到至少有一个通信可以进行下去；

package mainfunc main() {ch := make(chan int, 1)go func(chan int) { // go func(ch chan int) { 这样写也可以？ 为啥？for {select {case ch <- 0:case ch <- 1:}}}(ch)for i := 0; i < 10; i++ {println(<-ch)}}

输出结果：

如果需要同时处理多个 channel ，可使用 select 语句。它随机选择一个可用 channel 做收发操作，或执行 default case 。

package mainimport ("fmt""os"
)func main() {a, b := make(chan int, 3), make(chan int)go func() {v, ok, s := 0, false, ""for {select { // 随机选择可⽤用 channel，接收数据。case v, ok = <-a:s = "a"case v, ok = <-b:s = "b"}if ok {fmt.Println(s, v)} else {os.Exit(0)}}}()for i := 0; i < 5; i++ {select { // 随机选择可用 channel，发送数据。case a <- i:case b <- i:}}close(a)select {} // 没有可用 channel，阻塞 main goroutine。
}

输出：

a 0
a 1
a 2
a 3
b 4

在循环中使用 select default case 需要小心，避免形成洪水。

如果在 select 语句中发现某个通道已关闭，那么应该怎样屏蔽掉它所在的分支？

在 case 中通过第二个参数判断 chan 是否关闭，如果关闭则通过 make(chan type) 来对关闭的 chan 置 nil ，当再次执行到 select 时，因为 chan 时 nil 会进入阻塞而不会进入候选分支。

package mainimport ("fmt""time"
)func main() {i := 0c := make(chan int, 2)c <- 1c <- 2close(c)for {select {case value, ok := <-c:if !ok {c = make(chan int)fmt.Println("ch is closed")} else {fmt.Printf("value is %#v\n", value)}default:time.Sleep(1e9) // 等待1秒钟fmt.Println("default, ", i)i = i + 1if i > 3 {return}}}
}

输出结果：

value is 1
value is 2
ch is closed
default,  0
default,  1
default,  2
default,  3

在 select 语句与 for 语句联用时，怎样直接退出外层的 for 语句？

可以使用 goto 加 lable 跳转到 for 外面；
可以设置一个额外的标记位，当 chan 关闭时，设置 flag=true ，在 for 的最后判断 flag 决定是否 break ；

5. 用 channel 实现信号量 (semaphore)

package mainimport ("fmt""sync"
)func main() {wg := sync.WaitGroup{}wg.Add(3)sem := make(chan int, 1)for i := 0; i < 3; i++ {go func(id int) {defer wg.Done()sem <- 1 // 向 sem 发送数据，阻塞或者成功。for x := 0; x < 3; x++ {fmt.Println(id, x)}<-sem // 接收数据，使得其他阻塞 goroutine 可以发送数据。}(i)}wg.Wait()
}

输出：

6. 用 closed channel 发出退出通知

close 函数声明如下：

func close(c chan<- Type)

内置的 close 函数，只能用于 chan 类型变量。使用 close 函数关闭通道后，这个通道不允许被写入新的信息，但是关闭操作不会清除通道中已有的内容，不影响通道被读取。示例代码如下：

package main
import ("fmt""time"
)
func write(ch chan int) {for i := 0; i < 10; i++ {ch <- i * 10time.Sleep(time.Second * 1)}close(ch)
}
func read(ch chan int) {for {if val, ok := <-ch; ok {fmt.Println("从通道中读取值：", val)} else {// 通道被关闭fmt.Println("通道已关闭，退出读取程序")break}}
}
func main() {var ch = make(chan int, 10)go write(ch)read(ch)
}

上边的通道读取操作是:

val,ok := <-ch

当通道被关闭后：

如果从通道中读取到信息，则 ok 值为 true ， val 是一个有效值；
如果从通道中没有读取到信息，则 ok 值为 false ，此时的 val 是脏数据，切勿将 ok 为 false 时的 val 值拿去使用，此时的 val 值是 chan 指定数据类型的默认值。

如果通道没有被关闭，当从通道中没有读取到信息时，读取操作将会产生程序阻塞。所以使用 close 函数的目的是关闭不会再写入数据的通道，告诉通道读取方，所有数据发送完毕。

package mainimport ("sync""time"
)func main() {var wg sync.WaitGroupquit := make(chan bool)for i := 0; i < 2; i++ {wg.Add(1)go func(id int) {defer wg.Done()task := func() {println(id, time.Now().Nanosecond())time.Sleep(time.Second)}for {select {case <-quit: // closed channel 不会阻塞，因此可用作退出通知。returndefault: // 执行正常任务。task()}}}(i)}time.Sleep(time.Second * 5) // 让测试 goroutine 运行一会。close(quit)                 // 发出退出通知。wg.Wait()
}

7. channel 传参或者作为结构成员

channel 是第一类对象，可传参 (内部实现为指针) 或者作为结构成员。

package mainimport "fmt"type Request struct {data []intret  chan int
}func NewRequest(data ...int) *Request {return &Request{data, make(chan int, 1)}
}
func Process(req *Request) {x := 0for _, i := range req.data {x += i}req.ret <- x
}
func main() {req := NewRequest(10, 20, 30)Process(req)fmt.Println(<-req.ret)
}

8. 并发总结

并发是指 goroutine 运行的时候是相互独立的。
使用关键字 go 创建 goroutine 来运行函数。
goroutine 在逻辑处理器上执行，而逻辑处理器具有独立的系统线程和运行队列。
竞争状态是指两个或者多个 goroutine 试图访问同一个资源。
原子函数和互斥锁提供了一种防止出现竞争状态的办法。
通道提供了一种在两个 goroutine 之间共享数据的简单方法。
无缓冲的通道保证同时交换数据，而有缓冲的通道不做这种保证。