如何设置最大数量的 goroutine 来完成工作？答案

【问题标题】：How to set a max number of goroutines to complete the work?如何设置最大数量的 goroutine 来完成工作？
【发布时间】：2020-12-14 03:51:55
【问题描述】：

var wg sync.WaitGroup
wg.Add(len(work))

sem := make(chan struct{}, 10)

wgDone := make(chan bool)

for i < len(work)-1 {
    go func() {
        defer wg.Done()
        sem <- struct{}{}
        defer func() {
            <-sem
        }()
        worker(work[i])
    }()
    i = i + 1
}

go func() {     
    wg.Wait()
    close(wgDone)
}()

我一次只需要 10 个新的 goroutines 来执行这项工作。这是我目前的解决方案，它阻止 goroutines 继续，所以一次只有 10 个。我该如何改变它，这样它就不会创建大量阻塞等待工作的 goroutine，而是只创建 10 个完成所有工作的 goroutine？

【问题讨论】：

问题是什么？

标签： go goroutine

【解决方案1】：

根据用例，其中一种方法很有用：

使用max 的新 goroutine 数量和一个通道作为队列 (The Go playground)：

package main

import (
    "fmt"
    "sync"
)

func main() {
    const max = 10
    queue := make(chan int, max)
    wg := &sync.WaitGroup{}
    for i := 0; i < max; i++ {
        wg.Add(1)
        go worker(wg, queue)
    }
    for i := 0; i < 100; i++ {
        queue <- i
    }
    close(queue)
    wg.Wait()
    fmt.Println("Done")
}
func worker(wg *sync.WaitGroup, queue chan int) {
    defer wg.Done()
    for job := range queue {
        fmt.Print(job, " ") // a job
    }
}

使用缓冲通道作为信号量来限制 新的 goroutine 数量 到 max 数量 (The Go playground)：

package main

import (
    "fmt"
    "sync"
)

func main() {
    const max = 10
    semaphore := make(chan struct{}, max)
    wg := &sync.WaitGroup{}
    for i := 0; i < 1000; i++ {
        semaphore <- struct{}{} // acquire
        wg.Add(1)
        go limited(i, wg, semaphore)
    }
    wg.Wait()
    fmt.Println("Done")
}
func limited(i int, wg *sync.WaitGroup, semaphore chan struct{}) {
    defer wg.Done()
    fmt.Println("i =", i) // a job
    <-semaphore // release
}

使用缓冲通道作为信号量来限制 jobs 到最大数量 - 这里的 goroutine 数量超过最大数量 (The Go playground)：

package main

import (
    "fmt"
    "sync"
)

func main() {
    const max = 10
    semaphore := make(chan struct{}, max)
    wg := &sync.WaitGroup{}
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go limited(i, wg, semaphore)
    }
    wg.Wait()
    fmt.Println("Done")
}
func limited(i int, wg *sync.WaitGroup, semaphore chan struct{}) {
    defer wg.Done()
    semaphore <- struct{}{} // acquire
    fmt.Println("i =", i) // a job
    <-semaphore // release
}

【讨论】：

【解决方案2】：

因此，如果您只想说 10 个工作人员，则应该生成 10 个工作人员来监听作业队列，这可以是一个通道，您可以将输入推送到该通道，工作人员会选择它

现在它只会在队列已满时阻塞作业，因此您可以根据您的用例决定队列大小

package main

import (
    "fmt"
    "sync"
)

var jobQ chan int
var wg sync.WaitGroup

func main() {
    jobQ = make(chan int, 100)
    
    go func(){
        wg.Add(1)
        defer wg.Done() 
        //Spawn 10 workers
        for i:=0;i<10;i++ {
            fmt.Println("Spawn :", i)
            wg.Add(1)
            go worker(jobQ)
        }
    }()
    
    for i := 0; i< 1000;i++ {
        jobQ<- i
    }
    close(jobQ)
    
    
    wg.Wait()
}



func worker(jobs chan int) {
    defer wg.Done()
    for job:=range jobs {
        fmt.Println(job)
    }
}

现在您可以自定义它并找到其他工作池实现；工作池被大量使用，你会发现不同的实现

游乐场：https://play.golang.org/p/lzIMRUCvqR9

【讨论】：