使用 channel 控制并发数量
1. 协程
goroutine 是轻量级线程,调度由 Go 运行时进行管理的。Go 语言的并发控制主要使用关键字 go 开启协程 goroutine。Go 协程(Goroutine)之间通过信道(channel)进行通信,简单的说就是多个协程之间通信的管道。信道可以防止多个协程访问共享内存时发生资源争抢的问题。语法格式:
1// 普通函数创建 goroutine
2go 函数名(参数列表)
3
4//匿名函数创建 goroutine
5go func(参数列表){
6 //函数体
7}(调用参数列表)
协程可以开启多少个?是否有限制呢?
1func testRoutine() {
2 var wg sync.WaitGroup
3 for i := 0; i < math.MaxInt32; i++ {
4 wg.Add(1)
5 go func(i int) {
6 defer wg.Done()
7 fmt.Printf("并发数量:%d/n", i)
8 time.Sleep(time.Second)
9 }(i)
10 }
11 wg.Wait()
12}
以上代码开启了 math.MaxInt32个协程的并发,执行后可以看到结果直接 panic:“panic: too many concurrent operations on a single file or socket (max 1048575)”。整个并发操作超出了系统最大值。
2. 控制协程数量
sync 同步机制
使用 sync.WaitGroup 启动指定数量的协程 goroutine。
1func testRoutine() {
2 var wg = sync.WaitGroup{}
3
4 taskCount := 5 // 指定并发数量
5 for i := 0; i < taskCount; i++ {
6 wg.Add(1)
7 go func(i int) {
8 fmt.Println("go func ", i)
9 wg.Done()
10 }(i)
11 }
12 wg.Wait()
13}
如果 taskcount 设置的很大超出了限制的,则其还是没有控制到并发数量。可以优化下设计,类似池的设计思想,通过允许最大连接数控制量,当超出了数量就需要等待释放,有空闲的连接的时候才可以继续执行。
1func testRoutine() {
2 task_chan := make(chan bool, 3) //100 为 channel长度
3 wg := sync.WaitGroup{}
4 defer close(task_chan)
5 for i := 0; i < math.MaxInt; i++ {
6 wg.Add(1)
7 fmt.Println("go func ", i)
8 task_chan <- true
9 go func() {
10 <-task_chan
11 defer wg.Done()
12 }()
13 }
14
15 wg.Wait()
16}
- 创建缓冲区大小为 3 的 channel,在没有被接收的情况下,至多发送 3 个消息则被阻塞。通过 channel 控制每次并发的数量。
- 开启协程前,设置 task_chan <- true,若缓存区满了则阻塞
- 协程任务执行完成后就释放缓冲区
- 等待所有的并发都处理结束后则函数结束。其实可以不使用 sync.WaitGroup。因使用 channel 控制并发处理的任务数量可以不用使用等待并发处理结束。
3. 参考资料:
boilingfrog.github.io/2021/04/14/控制goroutine的数量/