concurrency - 使用固定数量的工作人员处理永无休止的队列中的作业

Question

这让我很头疼，我不知道如何解决；

• 我想要有固定数量的 N 个 goroutine 并行运行
• 我将从永无休止的队列中获取 X 条有关要处理的作业的消息
• 我想让 N 个 goroutine 处理这 X 个作业，一旦其中一个例程无事可做，我就想从永无休止的队列中获取另外 X 个作业

下面答案中的代码(参见网址)可以出色地处理任务，但是一旦任务列表为空，工作人员就会死亡，我希望他们保持活力并以某种方式通知主代码他们失业了这样我就可以获取更多工作来填充任务列表

How would you define a pool of goroutines to be executed at once in Golang?

使用下面的 user:Jsor 示例代码，我尝试创建一个简单的程序，但我很困惑。

import (
    "fmt"
    "strconv"
)

//workChan - read only that delivers work
//requestChan - ??? what is this
func Worker(myid string, workChan <- chan string, requestChan chan<- struct{}) {
    for {
        select {
        case work := <-workChan:
            fmt.Println("Channel: " + myid + " do some work: " + work)
        case requestChan <- struct{}{}:
            //hm? how is the requestChan used?
        }
    }
}

func Logic(){

    workChan := make(chan string)
    requestChan := make(chan struct{})

    //Create the workers
    for i:=1; i < 5; i++ {
        Worker( strconv.Itoa( i), workChan, requestChan)
    }

    //Give the workers some work
    for i:=100; i < 115; i++ {
        workChan<- "workid"+strconv.Itoa( i)
    }

}

Answer 1

这就是 select 语句的用途。

func Worker(workChan chan<- Work, requestChan chan<- struct{}) {
    for {
        select {
        case work := <-workChan:
            // Do work
        case requestChan <- struct{}{}:
        }
    }
}

这个 worker 将永远运行。如果工作可用，它将从工作 channel 中提取。如果没有剩余，它将发送请求。

并不是说它会永远运行，如果你想杀死一个 worker ，你需要做其他事情。一种可能性是始终使用workChan 检查ok，如果该 channel 已关闭，则退出该函数。另一种选择是为每个工作人员使用单独的退出 channel 。