我正在尝试在 Go 中实现一个工人池。go-wiki （以及 Channels 部分中的Effective Go）提供了边界资源使用的优秀示例。只需使用与工作池一样大的缓冲区创建一个通道。然后用工人填充该通道，并在完成后将它们发送回通道。从通道阻塞接收，直到有工作人员可用。所以通道和循环是整个实现——非常酷！
或者，可以阻止发送到通道，但想法相同。
我的问题是关于在运行时更改工作池的大小。我不相信有办法改变频道的大小。我有一些想法，但大多数似乎都太复杂了。 这个页面实际上以几乎相同的方式使用通道和空结构实现信号量，但它有同样的问题（这些事情在谷歌搜索“golang semaphore”时一直出现。
我会反过来做。我不会生成许多 goroutine（仍然需要大量内存）并使用通道来阻止它们，而是将工作程序建模为 goroutine 并使用通道来分配工作。像这样的东西：
package main
import (
"fmt"
"sync"
)
type Task string
func worker(tasks <-chan Task, quit <-chan bool, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case task, ok := <-tasks:
if !ok {
return
}
fmt.Println("processing task", task)
case <-quit:
return
}
}
}
func main() {
tasks := make(chan Task, 128)
quit := make(chan bool)
var wg sync.WaitGroup
// spawn 5 workers
for i := 0; i < 5; i++ {
wg.Add(1)
go worker(tasks, quit, &wg)
}
// distribute some tasks
tasks <- Task("foo")
tasks <- Task("bar")
// remove two workers
quit <- true
quit <- true
// add three more workers
for i := 0; i < 3; i++ {
wg.Add(1)
go worker(tasks, quit, &wg)
}
// distribute more tasks
for i := 0; i < 20; i++ {
tasks <- Task(fmt.Sprintf("additional_%d", i+1))
}
// end of tasks. the workers should quit afterwards
close(tasks)
// use "close(quit)", if you do not want to wait for the remaining tasks
// wait for all workers to shut down properly
wg.Wait()
}
使用一些方便的方法创建一个单独的 WorkerPool 类型可能是个好主意。此外，使用还包含用于指示任务已成功执行type Task string的通道的结构不是很常见的。done
编辑：我玩了更多，想出了以下内容：http ://play.golang.org/p/VlEirPRk8V 。它基本上是相同的示例，但具有更好的 API。
可以想到的一个简单的改变是有一个通道来控制信号量的大小。相关部分是选择语句。如果队列中有更多工作，则使用当前信号量处理它。如果有更改信号量大小的请求，请更改它并使用新信号量继续处理 req 队列。请注意，旧的将被垃圾收集。
package main
import "time"
import "fmt"
type Request struct{ num int }
var quit chan struct{} = make(chan struct{})
func Serve(queue chan *Request, resize chan int, semsize int) {
for {
sem := make(chan struct{}, semsize)
var req *Request
select {
case semsize = <-resize:
{
sem = make(chan struct{}, semsize)
fmt.Println("changing semaphore size to ", semsize)
}
case req = <-queue:
{
sem <- struct{}{} // Block until there's capacity to process a request.
go handle(req, sem) // Don't wait for handle to finish.
}
case <-quit:
return
}
}
}
func process(r *Request) {
fmt.Println("Handled Request", r.num)
}
func handle(r *Request, sem chan struct{}) {
process(r) // May take a long time & use a lot of memory or CPU
<-sem // Done; enable next request to run.
}
func main() {
workq := make(chan *Request, 1)
ctrlq := make(chan int)
go func() {
for i := 0; i < 20; i += 1 {
<-time.After(100 * time.Millisecond)
workq <- &Request{i}
}
<-time.After(500 * time.Millisecond)
quit <- struct{}{}
}()
go func() {
<-time.After(500 * time.Millisecond)
ctrlq <- 10
}()
Serve(workq, ctrlq, 1)
}