11

我正在尝试在 Go 中实现一个工人池。go-wiki （以及 Channels 部分中的Effective Go）提供了边界资源使用的优秀示例。只需使用与工作池一样大的缓冲区创建一个通道。然后用工人填充该通道，并在完成后将它们发送回通道。从通道阻塞接收，直到有工作人员可用。所以通道和循环是整个实现——非常酷！

或者，可以阻止发送到通道，但想法相同。

我的问题是关于在运行时更改工作池的大小。我不相信有办法改变频道的大小。我有一些想法，但大多数似乎都太复杂了。 这个页面实际上以几乎相同的方式使用通道和空结构实现信号量，但它有同样的问题（这些事情在谷歌搜索“golang semaphore”时一直出现。

4

2 回答 2

25

我会反过来做。我不会生成许多 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。

于 2014-05-23T21:47:58.403 回答
3

可以想到的一个简单的改变是有一个通道来控制信号量的大小。相关部分是选择语句。如果队列中有更多工作，则使用当前信号量处理它。如果有更改信号量大小的请求，请更改它并使用新信号量继续处理 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)
}

http://play.golang.org/p/AHOLlAv2LH

于 2014-05-23T21:08:44.920 回答