如何关闭多个 goroutine 正在发送的通道？

Question

我正在尝试并行进行一些计算。该程序的设计使得每个工作程序 goroutine 将已解决难题的“碎片”发送回控制器 goroutine，后者等待接收并组装从工作程序发送的所有内容。关闭单通道的 idomatic Go 是什么？我无法在每个 goroutine 中的通道上调用 close，因为那样我可能会在关闭的通道上发送。同样，无法预先确定哪个 goroutine 将首先完成。这里需要 sync.WaitGroup 吗？

Answer 1

这是一个使用sync.WaitGroup来做你正在寻找的事情的例子，

这个例子接受一个很长的整数列表，然后通过向 N 个并行工作人员处理一个相等大小的输入数据块来将它们全部相加。它可以在go playground上运行：

package main

import (

    "fmt"

    "sync"

)

const WorkerCount = 10

func main() {

    // Some input data to operate on.

    // Each worker gets an equal share to work on.

    data := make([]int, WorkerCount*10)

    for i := range data {

        data[i] = i

    }

    // Sum all the entries.

    result := sum(data)

    fmt.Printf("Sum: %d\n", result)

}

// sum adds up the numbers in the given list, by having the operation delegated

// to workers operating in parallel on sub-slices of the input data.

func sum(data []int) int {

    var sum int

    result := make(chan int)

    defer close(result)

    // Accumulate results from workers.

    go func() {

        for {

            select {

            case value := <-result:

                sum += value

            }

        }

    }()

    // The WaitGroup will track completion of all our workers.

    wg := new(sync.WaitGroup)

    wg.Add(WorkerCount)

    // Divide the work up over the number of workers.

    chunkSize := len(data) / WorkerCount

    // Spawn workers.

    for i := 0; i < WorkerCount; i++ {

        go func(i int) {

            offset := i * chunkSize

            worker(result, data[offset:offset+chunkSize])

            wg.Done()

        }(i)

    }

    // Wait for all workers to finish, before returning the result.

    wg.Wait()

    return sum

}

// worker sums up the numbers in the given list.

func worker(result chan int, data []int) {

    var sum int

    for _, v := range data {

        sum += v

    }

    result <- sum

}

Answer 2

是的，这是 sync.WaitGroup 的完美用例。

您的另一种选择是每个 goroutine 使用 1 个通道，并使用一个多路复用器 goroutine，从每个通道输入单个通道。但这会很快变得笨拙，所以我只需要一个sync.WaitGroup。