如何从按特定顺序执行的 N 个 goroutine 中收集值？

Question

下面是 Stuff 类型的结构。它有三个整数。A Number，它的Double和它的Power。让我们假设计算给定整数列表的 double 和 power 是一项昂贵的计算。

type Stuff struct {

    Number int

    Double int

    Power  int

}

func main() {

    nums := []int{2, 3, 4} // given numbers

    stuff := []Stuff{}     // struct of stuff with transformed ints

    double := make(chan int)

    power := make(chan int)

    for _, i := range nums {

        go doubleNumber(i, double)

        go powerNumber(i, power)

    }

    // How do I get the values back in the right order?

    fmt.Println(stuff)

}

func doubleNumber(i int, c chan int) {

    c <- i + i

}

func powerNumber(i int, c chan int) {

    c <- i * i

}

的结果fmt.Println(stuff)应该与初始化内容相同，例如：

stuff := []Stuff{

    {Number: 2, Double: 4, Power: 4}

    {Number: 3, Double: 6, Power: 9}

    {Number: 4, Double: 8, Power: 16}

}

我知道我可以使用<- double并<- power从通道中收集值，但我不知道什么双/幂属于什么数字。

Answer 1

Goroutines 并发、独立地运行，因此如果没有显式同步，您将无法预测执行和完成顺序。因此，您不能将返回的数字与输入的数字配对。您可以返回更多数据（例如，输入数字和输出，例如包装在一个结构中），或者将指针传递给工作函数（作为新的 goroutines 启动），例如让 goroutines 在其自身*Stuff中填充计算数据。Stuff返回更多数据我将使用一个通道类型chan Pair，其中Pair是：type Pair struct{ Number, Result int }所以计算将如下所示：func doubleNumber(i int, c chan Pair) { c <- Pair{i, i + i} }func powerNumber(i int, c chan Pair) { c <- Pair{i, i * i} }我将使用 amap[int]*Stuff因为可收集的数据来自多个渠道（double和power），并且我想Stuff轻松快速地找到合适的（需要指针，所以我也可以“在地图中”修改它）。所以主要功能：nums := []int{2, 3, 4} // given numbersstuffs := map[int]*Stuff{}double := make(chan Pair)power := make(chan Pair)for _, i := range nums {&nbsp; &nbsp; go doubleNumber(i, double)&nbsp; &nbsp; go powerNumber(i, power)}// How do I get the values back in the right order?for i := 0; i < len(nums)*2; i++ {&nbsp; &nbsp; getStuff := func(number int) *Stuff {&nbsp; &nbsp; &nbsp; &nbsp; s := stuffs[number]&nbsp; &nbsp; &nbsp; &nbsp; if s == nil {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; s = &Stuff{Number: number}&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; stuffs[number] = s&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; return s&nbsp; &nbsp; }&nbsp; &nbsp; select {&nbsp; &nbsp; case p := <-double:&nbsp; &nbsp; &nbsp; &nbsp; getStuff(p.Number).Double = p.Result&nbsp; &nbsp; case p := <-power:&nbsp; &nbsp; &nbsp; &nbsp; getStuff(p.Number).Power = p.Result&nbsp; &nbsp; }}for _, v := range nums {&nbsp; &nbsp; fmt.Printf("%+v\n", stuffs[v])}输出（在Go Playground上试试）：&{Number:2 Double:4 Power:4}&{Number:3 Double:6 Power:9}&{Number:4 Double:8 Power:16}使用指针由于现在我们正在传递值，我们可以在其本身*Stuff中“预填充”输入数字。Stuff但必须小心，您只能通过适当的同步读取/写入值。最简单的方法是等待所有“worker” goroutine 完成它们的工作。var wg = &sync.WaitGroup{}func main() {&nbsp; &nbsp; nums := []int{2, 3, 4} // given numbers&nbsp; &nbsp; stuffs := make([]Stuff, len(nums))&nbsp; &nbsp; for i, n := range nums {&nbsp; &nbsp; &nbsp; &nbsp; stuffs[i].Number = n&nbsp; &nbsp; &nbsp; &nbsp; wg.Add(2)&nbsp; &nbsp; &nbsp; &nbsp; go doubleNumber(&stuffs[i])&nbsp; &nbsp; &nbsp; &nbsp; go powerNumber(&stuffs[i])&nbsp; &nbsp; }&nbsp; &nbsp; wg.Wait()&nbsp; &nbsp; fmt.Printf("%+v", stuffs)}func doubleNumber(s *Stuff) {&nbsp; &nbsp; defer wg.Done()&nbsp; &nbsp; s.Double = s.Number + s.Number}func powerNumber(s *Stuff) {&nbsp; &nbsp; defer wg.Done()&nbsp; &nbsp; s.Power = s.Number * s.Number}输出（在Go Playground上试试）：[{Number:2 Double:4 Power:4} {Number:3 Double:6 Power:9} {Number:4 Double:8 Power:16}]同时写入不同的切片元素另请注意，由于您可以同时编写不同的数组或切片元素（详细信息请参见我可以同时编写不同的切片元素），因此您可以将结果直接写入没有通道的切片中。请参阅重构代码以惯用的方式使用单个通道。

Answer 2

就个人而言，我会使用 achan Stuff将结果传回，然后启动 goroutines 计算一个完整的Stuff并将其传回。如果您需要同时计算的单个部分的各个部分Stuff，您可以使用专用通道从每个 goroutine 生成 goroutine。收集完所有结果后，您可以（可选）使用累积值对切片进行排序。我在下面的意思示例（原则上，您可以使用 async.WaitGroup来协调事物，但如果输入计数已知，则严格来说不需要它）。type Stuff struct {&nbsp; number int64&nbsp; double int64&nbsp; square int64}// Compute a Stuff with individual computations in-line, send it outfunc computeStuff(n int64, out chan<- Stuff) {&nbsp; &nbsp;rv := Stuff{number: n}&nbsp; &nbsp;rv.double = n * 2&nbsp; &nbsp;rv.square = n * n&nbsp; &nbsp;out <- rv}// Compute a Stuff with individual computations concurrentfunc computeStuffConcurrent(n int64, out chan<- Stuff) {&nbsp; rv := Stuff{number: n}&nbsp; dc := make(chan int64)&nbsp; sc := make(chan int64)&nbsp; defer close(dc)&nbsp; defer close(sc)&nbsp; go double(n, dc)&nbsp; go square(n, sc)&nbsp; rv.double = <-dc&nbsp; rv.square = <-sc&nbsp; out <- rv}func double(n int64, result chan<- int) {&nbsp; &nbsp;result <- n * 2}func square(n int64, result chan<- int) {&nbsp; result <- n * n}func main() {&nbsp; inputs := []int64{1, 2, 3}&nbsp; results := []Stuff{}&nbsp; resultChannel := make(chan Stuff)&nbsp; for _, input := range inputs {&nbsp; &nbsp; go computeStuff(input, resultChannel)&nbsp;&nbsp; &nbsp; // Or the concurrent version, if the extra performance is needed&nbsp; }&nbsp; for c := 0; c < len(inputs); c++ {&nbsp; &nbsp; results = append(results, <- resultChannel)&nbsp; }&nbsp; // We now have all results, sort them if you need them sorted}

Answer 3

Ordered-concurrently 是一个 go 模块，它同时处理工作并按照提供的顺序返回数据。https://github.com/tejzpr/ordered-concurrently示例 - https://play.golang.org/p/hkcIuRHj63hpackage mainimport (&nbsp; &nbsp; concurrently "github.com/tejzpr/ordered-concurrently/v2"&nbsp; &nbsp; "log"&nbsp; &nbsp; "math/rand"&nbsp; &nbsp; "time")type loadWorker int// The work that needs to be performed// The input type should implement the WorkFunction interfacefunc (w loadWorker) Run() interface{} {&nbsp; &nbsp; time.Sleep(time.Millisecond * time.Duration(rand.Intn(10)))&nbsp; &nbsp; return w}func main() {&nbsp; &nbsp; max := 10&nbsp; &nbsp; inputChan := make(chan concurrently.WorkFunction)&nbsp; &nbsp; output := concurrently.Process(inputChan, &concurrently.Options{PoolSize: 10, OutChannelBuffer: 10})&nbsp; &nbsp; go func() {&nbsp; &nbsp; &nbsp; &nbsp; for work := 0; work < max; work++ {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; inputChan <- loadWorker(work)&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; close(inputChan)&nbsp; &nbsp; }()&nbsp; &nbsp; for out := range output {&nbsp; &nbsp; &nbsp; &nbsp; log.Println(out.Value)&nbsp; &nbsp; }}免责声明：我是模块创建者