能看到这篇文章一定是特殊的缘分,请务必珍惜,请详细看看吧,哈哈。
图片
不止上图,最近 Go就业训练营 中不少小伙伴说,面试中碰到了好几次让手撕协程池的公司。
通过以上的解题思路,我们可以实现一个基本的协程池。
在实际应用中,可能还需要考虑一些其他的因素,如任务优先级、任务超时处理等。根据具体的需求,可以对协程池进行进一步的扩展和优化。
说完了解题思路,再给大家一个可参考,可运行的示例代码:
package main
import (
"fmt"
"sync"
)
type Job struct {
ID int
}
type Worker struct {
ID int
JobChannel chan Job
Quit chan bool
}
type Pool struct {
WorkerNum int
JobChannel chan Job
WorkerQueue chan chan Job
Quit chan bool
wg sync.WaitGroup
}
// NewWorker 创建一个新的工作者
func NewWorker(id int, workerQueue chan chan Job) Worker {
return Worker{
ID: id,
JobChannel: make(chan Job),
Quit: make(chan bool),
}
}
// Start 启动工作者
func (w Worker) Start(workerQueue chan chan Job) {
go func() {
for {
workerQueue <- w.JobChannel
select {
case job := <-w.JobChannel:
fmt.Printf("Worker %d started job %d\n", w.ID, job.ID)
// 执行任务
fmt.Printf("Worker %d finished job %d\n", w.ID, job.ID)
case <-w.Quit:
return
}
}
}()
}
// Start 启动工作者池
func (p *Pool) Start() {
for i := 0; i < p.WorkerNum; i++ {
worker := NewWorker(i, p.WorkerQueue)
worker.Start(p.WorkerQueue)
}
go func() {
for {
select {
case job := <-p.JobChannel:
worker := <-p.WorkerQueue
worker <- job
case <-p.Quit:
for i := 0; i < p.WorkerNum; i++ {
worker := <-p.WorkerQueue
worker <- Job{} // 发送空任务,通知协程退出
}
p.wg.Done()
return
}
}
}()
}
// AddJob 添加作业到作业通道
func (p *Pool) AddJob(job Job) {
p.JobChannel <- job
}
// Stop 停止工作者池
func (p *Pool) Stop() {
p.Quit <- true
p.wg.Wait()
}
func main() {
pool := Pool{
WorkerNum: 5,
JobChannel: make(chan Job),
WorkerQueue: make(chan chan Job, 5),
Quit: make(chan bool),
}
pool.Start()
for i := 0; i < 10; i++ {
job := Job{ID: i}
pool.AddJob(job)
}
pool.Stop()
}
以下是对代码的注释:
有小伙伴提出了疑问:
WorkerQueue chan chan Job 的作用是什么?为什么要这么定义?
WorkerQueue chan chan Job 的作用是用于传递工作者(Worker)的作业通道(JobChannel)。它是一个通道(channel),其中每个元素都是一个作业通道。
为什么要这么定义呢?这是因为在工作者池模式中,每个工作者需要一个独立的作业通道来接收作业。通过将每个工作者的作业通道放入一个通道中,可以实现对工作者的动态分配和管理。
具体来说,WorkerQueue 通道用于存储每个工作者的作业通道。当有新的作业到达时,工作者池会从 WorkerQueue 中取出一个可用的工作者的作业通道,并将作业发送到该通道中,由相应的工作者进行处理。
这种设计可以有效地控制并发任务的分配和调度。通过将工作者的作业通道放入 WorkerQueue 中,可以实现对工作者的复用和动态管理,避免了频繁地创建和销毁工作者协程的开销。
总结起来,WorkerQueue 的定义允许工作者池动态地管理工作者的作业通道,实现对并发任务的高效分配和调度。
图片
以上代码实现了一个简单的工作池(Worker Pool)模式。工作池由一组固定数量的工作者(Worker)协程组成,它们从作业通道(JobChannel)中获取作业(Job)并执行。
本文转载自微信公众号「 程序员升级打怪之旅」,作者「 王中阳Go」,可以通过以下二维码关注。
转载本文请联系「 程序员升级打怪之旅」公众号。
下一篇:大型制造企业智能决策场景解析