同步示例:
type job struct {
Id int
Message string
}
for {
// getJob() blocks until job is received
job := getJob()
doSomethingWithJob(job)
}
我希望处理从
getJob
和doSomethingWithJob
传入的作业。例如getJob可以是从MessagingQueue(例如RabbitMQ/Beanstalkd)接收到的有效负载,也可以是处理HTTP请求的有效负载。当我是
getJob
时,我不想阻止doSomethingWithJob
,反之亦然。但是,我确实希望控制/缓冲作业的数量,以免使系统过载。例如最大并发数为5。Go例程的概念此刻使我感到困惑,因此,朝着正确方向的任何指示都将非常有助于我学习。
更新:感谢@JimB的帮助。为什么 worker 5总是接手工作?
jobCh := make(chan *job)
// Max 5 Workers
for i := 0; i < 5; i++ {
go func() {
for job := range jobCh {
time.Sleep(time.Second * time.Duration(rand.Intn(3)))
log.Println(i, string(job.Message))
}
}()
}
for {
job, err := getJob()
if err != nil {
log.Println("Closing Channel")
close(jobCh)
break
}
jobCh <- job
}
log.Println("Complete")
输出示例
2016/06/09 22:19:57 5 {"id":10692,"name":"Test Message"}
2016/06/09 22:19:57 5 {"id":10687,"name":"Test Message"}
2016/06/09 22:19:57 5 {"id":10699,"name":"Test Message"}
2016/06/09 22:19:57 5 {"id":10701,"name":"Test Message"}
2016/06/09 22:19:57 5 {"id":10703,"name":"Test Message"}
2016/06/09 22:19:57 5 {"id":10704,"name":"Test Message"}
最佳答案
您可以从一个 channel 开始读取5个goroutine,以调用doSomethingWithJob
。这样一来,就不会同时处理5个以上的作业。
jobCh := make(chan *job)
// start 5 workers to process jobs
for i := 0; i < 5; i++ {
go func() {
for job := range jobCh {
doSomethingWithJob(job)
}
}()
}
// send jobs to workers as fast as we can
for {
jobCh <- getJob()
}