前言

参考 陈剑煜的博客，纯笔记，转载请尊重原创。

为何要控制

Go 最大的优势就是超强的并发能力，但是一次性开启过多 goroutine 也会带来严重的问题。
在个人电脑上运行下面的程序就会导致程序卡死。

  1. package main
  3. import (
  4.     "fmt"    
  5.     "time"
  6.     "math"
  7. )
  9. func main() {
  10.     userCount := math.MaxInt64
  11.     for i := 0; i < userCount; i++ {
  12.         go func(i int){
  13.             time.Sleep(time.Second)                    // 睡眠模拟处理逻辑
  14.             fmt.Printf("goroutine %d done\n", i)    // 处理结束
  15.         }(i)
  16.     }
  17. }

假设 userCount 是传入的参数，大小不可预知，同时开启 userCount 个 goroutine 处理同一件事，看起来效率会很高，但是当出现极端情况时（userCount 很大时，如 math.MaxInt64）会出现如下问题：

  • CPU 使用率上涨
  • 内存 / 虚拟内存 占用上涨
  • 主协程崩溃。

如何控制

控制细分两种：

  1. 控制同时产生 goroutine 的数量
  2. 控制同时运行的 goroutine 的数量

两者只有细微的差别，下面将重点探讨第 2 种情况。

channel 方案

利用 channel 的阻塞

  1. package main
  3. import (
  4.     "fmt"    
  5.     "time"
  6. )
  8. func main() {
  9.     userCount := 10
  10.     ch := make(chan bool, 2)            // 管道容量为 2
  11.     for i := 0; i < userCount; i++ {
  12.         // ch <- true    // place 1
  13.         go work(ch, i)
  14.     }
  15.     // time.Sleep(time.Second)            // 主协程不等待
  16. }
  18. func work(ch chan bool, i int) {
  19.     ch <- true        // place 2
  20.     fmt.Printf("goroutine %d done\n", i)    // 处理结束
  21.     <- ch
  22. }

代码分析：

  • 主协程一次性开启了 10 个子协程，每个子协程运行前都会往管道里发送一个数据，结束后接收一个数据，因此任意时刻最多只有两个子协程在运行。
  • 注意到代码中往管道发送数据的两个 place
    • 若放在 place 1，则主协程任意时刻只能开启两个子协程，并且这两个子协程都在运行。
    • 若放在 place 2，则主协程一次性开启了 10 个子协程，但是任意时刻最多只有两个子协程在运行。

理论上会输出 10 行，但是运行该程序可能没有输出，也可能输出不足 10 行，原因是主协程在子协程执行完之前就终止了，子协程也随之终止（在这里让主协程等待 1 秒可以让全部子协程执行完毕）。
由此可见，纯 channel 不能很好地控制 goroutine 的并发数量，主要缺点是主协程不知道最后一个子协程什么时候结束，这时候我们可以引入 sync.WaitGroup 来完善。

channel + sync.WaitGroup 方案

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6.     "time"
  7. )
  9. var wg sync.WaitGroup
  11. func main() {
  12.     userCount := 10
  13.     ch := make(chan bool, 2)
  14.     for i := 0; i < userCount; i++ {    
  15.         wg.Add(1)
  16.         go work(ch, i)
  17.     }
  18.     wg.Wait()
  19. }
  21. func work(ch chan bool, i int) {
  22.     ch <- true
  23.     time.Sleep(time.Second)                    // 睡眠模拟处理逻辑
  24.     fmt.Printf("goroutine %d done\n", i)    // 处理结束
  25.     wg.Done()
  26.     <- ch
  27. }

运行程序输出

  1. goroutine 0 done
  2. goroutine 4 done
  3. // 等待一下
  4. goroutine 5 done
  5. goroutine 9 done
  6. // 等待一下
  7. goroutine 6 done
  8. goroutine 7 done
  9. // 等待一下
  10. goroutine 2 done
  11. goroutine 8 done
  12. // 等待一下
  13. goroutine 3 done
  14. goroutine 1 done

确实做到了任意时刻只有两个 goroutine 在处理业务逻辑。
但是很多业务场景需要更好的灵活性，至少允许改变并发运行 goroutine 的数量，通常灵活是相对的，一般会设置并发量的最大。

灵活的 channel + sync.WaitGroup 方案

  1. package main
  3. import (
  4.     "fmt"
  5.     "sync"
  6.     "time"
  7.     // "time"
  8. )
  10. var wg sync.WaitGroup
  12. func main() {
  13.     userCount := 10
  14.     ch := make(chan int, 5)        // 最多支持 5 个 goroutine 并发运行
  15.     for i := 0; i < userCount; i++ {    
  16.         wg.Add(1)
  17.         go work(ch, i)
  18.     }
  20.     for i := 0; i < 10; i++ {
  21.         // 任意时刻最多有 2 个 goroutine 并发运行
  22.         ch <- i
  23.         ch <- i
  24.         // ch <- i
  25.         time.Sleep(time.Second)
  26.     }
  27.     close(ch)    // 主协程主动关闭管道
  28.     wg.Wait()
  29. }
  31. func work(ch chan int, i int) {
  32.     defer wg.Done()
  33.     for data := range ch {
  34.         fmt.Printf("goroutine %d get data %d\n", i, data)
  35.     }
  36. }

代码分析：

  • 管道的容量为 5 可以认为最多支持 5 个 goroutine 并发运行，控制一次往管道里发送数据的数量，就可以控制并发量。
  • 主协程必须主动关闭管道，否则将导致子协程一直阻塞等待从管道接收数据，进而导致死锁。

这只是一个最简单的灵活控制并发量的方案，有一点 goroutine pool（协程池）的雏形，目前有很多第三方库实现了很完善的 goroutine pool

这些都是目前比较成熟的第三方库，是生成和管理 goroutine pool 的工具。