将来自 go 例程的多个结果聚合到单个数组中答案

【问题标题】：Aggregating multiple results from go routines into a single array将来自 go 例程的多个结果聚合到单个数组中
【发布时间】：2017-08-17 02:27:35
【问题描述】：

我有以下函数可以衍生出一定数量的 go 例程

func (r *Runner) Execute() {
    var wg sync.WaitGroup
    wg.Add(len(r.pipelines))
    for _, p := range r.pipelines {
        go executePipeline(p, &wg)
    }

    wg.Wait()

    errs := ....//contains list of errors reported by any/all go routines

}

我在想频道可能有一些方法，但我似乎无法弄清楚。

【问题讨论】：

创建一个大小为len(r.pipelines) 的切片，并让每个工作人员写入其对应的索引。
有一种模式可以将一些频道的结果合并到一个名为 Fan In 的频道中。您可以在该模式中使用该方法（不一定是模式本身）。

标签： go channels

【解决方案1】：

如果可以使 executePipeline retuen 出错，一种方法是使用互斥锁：

// ...
for _, p := range r.pipelines {
    go func(p pipelineType) {
        if err := executePipeline(p, &wg); err != nil {
            mu.Lock()
            errs = append(errs, err)
            mu.UnLock()
        }
    }(p)
}

要使用通道，您可以使用单独的 goroutine 来列出错误：

errCh := make(chan error)

go func() {
    for e := range errCh {
        errs = append(errs, e)
    }
}

并在Execute 函数中进行以下更改：

// ...
wg.Add(len(r.pipelines))
for _, p := range r.pipelines {
    go func(p pipelineType) {
        if err := executePipeline(p, &wg); err != nil {
            errCh <- err
        }
    }(p)
}

wg.Wait()
close(errCh)

如果 goroutine 的数量不高，您可以随时使用上面列出的 @zerkms 方法。

您始终可以在函数本身内进行上述更改，而不是从 executePipleline 返回错误并使用匿名函数包装器。

【讨论】：

【解决方案2】：

您可以按照@Kaveh Shahbazian 的建议使用频道：

func (r *Runner) Execute() {
    pipelineChan := makePipeline(r.pipelines)

    for cnt := 0; cnt < len(r.pipelines); cnt++{
        //recieve from channel
        p := <- pipelineChan
        //do something with the result 
    }
}

func makePipeline(pipelines []pipelineType) <-chan pipelineType{
    pipelineChan := make(chan pipelineType)

    go func(){
        for _, p := range pipelines {
           go func(p pipelineType){
              pipelineChan <- executePipeline(p)
           }(p)
        }
    }()
    return pipelineChan
}

请看这个例子：https://gist.github.com/steven-ferrer/9b2eeac3eed3f7667e8976f399d0b8ad

【讨论】：