引言
在 Go 语言中创建协程(Goroutine)的成本非常低,因此稍不注意就可能创建出大量的协程,一方面会造成资源不断增长负载变高,另一方面不容易控制这些协程的状态。
不过,“能力越大,越需要克制”。网络上已经存在一些讲控制 Goroutine 数目的文章,本文通过图示的方式再简单总结一下其基本理念,以便于记忆。
Goroutine 数量不受控示例
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
var wg sync.WaitGroup
func main() {
jobsCount := 10
for j := 0; j < jobsCount; j++ {
wg.Add(1)
go do(j)
fmt.Printf("index: %d,goroutine Num: %d \n", j, runtime.NumGoroutine())
}
fmt.Println("done!")
}
func do(i int) {
defer wg.Done()
fmt.Printf("hello %d!\n", i)
time.Sleep(time.Second * 2) // 刻意睡 2 秒钟,模拟耗时
}
上面的代码假设有 jobsCount 个任务,通过 for-range 给每个任务创建了一个 Goroutine。为了让主协程等待所有的子协程执行完毕后再退出,使用 sync.WaitGroup 监控所有协程的状态,从而保证主协程结束时所有的子协程已经退出。为了说明问题,上面的代码还输出了 runtime.NumGoroutine() 的值用以表征协程的数量。
运行上面的代码,可以得到类似下面的输出。从下面的输出中我们可以得到两点信息:① 协程的执行顺序是随机的(比如 hello 3 在 hello 6 后面出现);② 协程的数量递增,最后到了 11 个之多。
输出如下:
go run .\main.go
index: 0,goroutine Num: 2
index: 1,goroutine Num: 3
hello 0!
hello 1!
index: 2,goroutine Num: 4
hello 2!
index: 3,goroutine Num: 5
index: 4,goroutine Num: 6
index: 5,goroutine Num: 7
hello 3!
hello 4!
hello 5!
index: 6,goroutine Num: 8
index: 7,goroutine Num: 9
index: 8,goroutine Num: 10
index: 9,goroutine Num: 11
done!
Goroutine 数量不受控制的图示
假如 CPU 只有 两个 核,下图展示了为每个 job 创建一个 goroutine 的情况(换句话说,goroutine 的数量是不受控制的)。此种情况虽然生成了很多的 goroutine,但是 每个 CPU 核上同一时间只能执行一个 goroutine ;当 job 很多且生成了相应数目的 goroutine 后,会出现很多等待执行的 goroutine,从而造成资源上的浪费。
Goroutine 数量受控制示例
Goroutine 数量受到限制的图示
给每个 job 生成一个 goroutine 的方式显得粗暴了很多,那么可以通过什么样的方式控制 goroutine 的数目呢?其实上面的代码通过一个 for-range 循环完成了两件事情:①为每个 job 创建 goroutine;②把任务相关的标识传给相应的 goroutine 执行。为了控制 goroutine 的数目,可以通过 buffered channel 给每个 goroutine 传递任务相关的信息 或者 可以把上面的两个过程拆分开:a)先通过一个 for-range 循环创建指定数目的 goroutine,b)然后通过 channel/buffered channel 给每个 goroutine 传递任务相关的信息(这里的channel是否缓冲无所谓,主要用到的是 channel 的线程安全特性)。如下图所示。
示例代码
方式一
通过有阻塞的 buffer channel 来控制 goroutine 增长
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
var wg sync.WaitGroup
func main() {
jobsCount := 10
ch := make(chan bool, 2) //通过 channel 控制 goroutine 数量,防止无休止增长
for j := 0; j < jobsCount; j++ {
wg.Add(1)
ch <- true
go do(ch, j)
fmt.Printf("index: %d,goroutine Num: %d \n", j, runtime.NumGoroutine())
}
wg.Wait()
fmt.Println("done!")
}
func do(ch chan bool, i int) {
defer wg.Done()
fmt.Printf("hello %d!\n", i)
time.Sleep(time.Second * 2) // 刻意睡 2 秒钟,模拟耗时
<-ch //必须在任务完成后从channel取出,通过channel来阻塞
}
方式二:
通过拆任务实现:
- a)先通过一个 for-range 循环创建指定数目的 goroutine
- b)然后通过 channel/buffered channel 给每个 goroutine 传递任务相关的信息
package main
import (
"fmt"
"runtime"
"sync"
"time"
)
var wg sync.WaitGroup
func main() {
jobsCount := 10
var jobsChan = make(chan int)
// a) 生成指定数目的 goroutine,每个 goroutine 消费 jobsChan 中的数据
poolCount := 2
for i := 0; i < poolCount; i++ {
go job(jobsChan)
}
// b) 把 job 依次推送到 jobsChan 供 goroutine 消费
for i := 0; i < jobsCount; i++ {
jobsChan <- i
wg.Add(1)
fmt.Printf("index: %d,goroutine Num: %d\n", i, runtime.NumGoroutine())
}
wg.Wait()
fmt.Println("done!")
}
func job(jobsChan chan int) {
for j := range jobsChan {
fmt.Printf("hello %d\n", j)
time.Sleep(time.Second * 2) // 刻意睡 2 秒钟,模拟耗时
wg.Done()
}
}
运行上面的代码可以得到下面类似的输出(可以看到 goroutine 的数量控制在了 3 个)
hello 0!
index: 0,goroutine Num: 2
index: 1,goroutine Num: 3
hello 1!
hello 2!
index: 2,goroutine Num: 3
index: 3,goroutine Num: 3
hello 3!
index: 4,goroutine Num: 3
hello 4!
index: 5,goroutine Num: 3
hello 5!
index: 6,goroutine Num: 3
index: 7,goroutine Num: 3
hello 7!
hello 6!
index: 8,goroutine Num: 3
index: 9,goroutine Num: 3
hello 8!
hello 9!
done!