导读
Channel
是Golang实现并发编程非常重要的组成部分,Channel
是一种内建的核心数据类型,需要使用make函数初始化,包括无缓冲的Channel(unbuffered Channel) 和有缓冲的Channel(buffered Channel)两种。无缓冲的Channel(unbuffered Channel) 主要用于goroutine之间的同步,有缓冲的Channel(buffered Channel)主要用于异步通信、控制goroutine并发数量。
Unbuffered := make(chan int) // Unbuffered channel of integer type
buffered := make(chan int, 10) // Buffered channel of integer type
场景
在我们的日常开发工作中,时常有需要控制并发的场景,如控制访问一个接口的并发,运维系统初始化机器服务的数量等。下面介绍一下如何使用有缓冲的Channel(buffered Channel)实现控制并发数量。
package main
import (
"flag"
"fmt"
"time"
)
// Fake a long and difficult work.
func DoWork() {
time.Sleep(5000 * time.Millisecond)
}
func main() {
maxNbConcurrentGoroutines := flag.Int("maxNbConcurrentGoroutines", 5, "the number of goroutines that are allowed to run concurrently")
nbJobs := flag.Int("nbJobs", 100, "the number of jobs that we need to do")
flag.Parse()
// Dummy channel to coordinate the number of concurrent goroutines.
// This channel should be buffered otherwise we will be immediately blocked
// when trying to fill it.
concurrentGoroutines := make(chan struct{}, *maxNbConcurrentGoroutines)
// The done channel indicates when a single goroutine has
// finished its job.
done := make(chan bool)
// The waitForAllJobs channel allows the main program
// to wait until we have indeed done all the jobs.
waitForAllJobs := make(chan bool)
// Collect all the jobs, and since the job is finished, we can
// release another spot for a goroutine.
go func() {
for i := 0; i < *nbJobs; i++ {
<-done
}
// We have collected all the jobs, the program
// can now terminate
waitForAllJobs <- true
}()
// Try to start nbJobs jobs
for i := 1; i <= *nbJobs; i++ {
fmt.Printf("ID: %v: waiting to launch!\n", i)
// Try to receive from the concurrentGoroutines channel. When we have something,
// it means we can start a new goroutine because another one finished.
// Otherwise, it will block the execution until an execution
// spot is available.
concurrentGoroutines <- struct{}{}
fmt.Printf("ID: %v: it's my turn!\n", i)
go func(id int) {
DoWork()
fmt.Printf("ID: %v: all done!\n", id)
done <- true
<-concurrentGoroutines
}(i)
}
// Wait for all jobs to finish
<-waitForAllJobs
}
这里代码源自Github,我对其做了一些改动,让执行流程更容易理解、输出结果更加明显,下面解释一下这段程序:
- 使用
flag
包创建命令行参数,可以自己指定并发的数量maxNbConcurrentGoroutines
和总共任务数量nbJobs
maxNbConcurrentGoroutines := flag.Int("maxNbConcurrentGoroutines", 5, "the number of goroutines that are allowed to run concurrently")
nbJobs := flag.Int("nbJobs", 100, "the number of jobs that we need to do")
flag.Parse()
运行一个并发为10,总数为100的示例:
go run limit_concurrency.go -maxNbConcurrentGoroutines 10 -nbJobs 100
- 创建
concurrentGoroutines
作为控制并发的带缓冲的Channel,done
用于记录一个goroutine完成,waitForAllJobs
用于阻塞main函数,等待所有goroutine完成。
concurrentGoroutines := make(chan struct{}, *maxNbConcurrentGoroutines)
done := make(chan bool)
waitForAllJobs := make(chan bool)
- 创建一个收集所有goroutine运行状态的goroutine,当所有goroutine完成后向
waitForAllJobs
发送“完成信号”。
go func() {
for i := 0; i < *nbJobs; i++ {
<-done
}
waitForAllJobs <- true
}()
- 这里我们叫它任务创建着,或者叫生成者也可以,首先会向
concurrentGoroutines
写入空的struct,因为concurrentGoroutines
的buffer是10,所以这里不会阻塞,直到for循环执行10次,将buffer填满,同时的也创建了10个goroutine用于执行我们的任务,当任务执行完毕(这里都暂定执行5s),向done
发送true通知第三步的goroutine我执行完了,接着读取concurrentGoroutines
,“释放”一个空间,让其他goroutine可以进来继续执行,但是怎么都不会超出buffer
的个数,等所有任务执行完,waitForAllJobs
收到了第三步gorotine发送的信号,整个程序结束,这就实现了控制并发。
for i := 1; i <= *nbJobs; i++ {
fmt.Printf("ID: %v: waiting to launch!\n", i)
concurrentGoroutines <- struct{}{}
fmt.Printf("ID: %v: it's my turn!\n", i)
go func(id int) {
DoWork()
fmt.Printf("ID: %v: all done!\n", id)
done <- true
<-concurrentGoroutines
}(i)
}
<-waitForAllJobs
好了,小伙伴们,快运行下看看结果吧,Let's ready to Go :)
有疑问加站长微信联系(非本文作者)