如何使用sync/errgroup包在Go中编写并发for循环

Question

我想同时对切片的元素执行操作
我正在使用sync/errgroup包来处理并发

这是 Go Playground 上的最小复制品https://go.dev/play/p/yBCiy8UW_80

import (
    "fmt"
    "golang.org/x/sync/errgroup"
)

func main() {
    eg := errgroup.Group{}
    input := []int{0, 1, 2}
    output1 := []int{}
    output2 := make([]int, len(input))
    for i, n := range input {
        eg.Go(func() (err error) {
            output1 = append(output1, n+1)
            output2[i] = n + 1
            return nil
        })
    }
    eg.Wait()
    fmt.Printf("with append %+v", output1)
    fmt.Println()
    fmt.Printf("with make %+v", output2)
}

输出

with append [3 3 3]
with make [0 0 3]

与预期相比[1 2 3]

Answer 1

这里有两个不同的问题：

首先，循环中的变量在每个 goroutine 有机会读取它们之前就会发生变化。当你有一个像这样的循环时

for i, n, := range input {
  // ...
}

变量i并n在整个循环期间有效。当控制到达循环底部并跳回顶部时，这些变量将被分配新值。如果循环中启动的 goroutine 正在使用这些变量，那么它们的值将发生不可预测的变化。这就是为什么您会看到相同的数字在示例的输出中多次出现。在第一次循环迭代中启动的 goroutine 直到n被设置为 2 后才开始执行。

为了解决这个问题，您可以执行 NotX 的答案所示的操作，并创建范围仅限于循环的单次迭代的新变量：

for i, n := range input {
  ic, nc := i, n
  // use ic and nc instead of i and n
}

循环内声明的变量的作用域仅限于循环的一次迭代，因此当循环的下一次迭代开始时，将创建全新的变量，从而防止原始变量在 goroutine 启动和实际开始运行之间发生更改。

其次，您同时从不同的 goroutine 修改相同的值，这是不安全的。特别是，您要append同时附加到同一个切片。在这种情况下发生的事情是不确定的，各种糟糕的事情都可能发生。

有两种方法可以解决这个问题。您已经设置的第一个：预先分配一个输出切片make，然后让每个 goroutine 填充切片中的特定位置：

output := make([]int, 3)
for i, n := range input {
  ic, nc := i, n
  eg.Go(func() (err error) {
    output[ic] = nc + 1
    return nil
  })
}
eg.Wait()

如果您知道启动循环时将有多少个输出，那么这会非常有效。

另一种选择是使用某种锁定来控制对输出切片的访问。sync.Mutex非常适合这个：

var output []int
mu sync.Mutex
for _, n := range input {
  nc := n
  eg.Go(func() (err error) {
    mu.Lock()
    defer mu.Unlock()
    output = append(output, nc+1)
    return nil
  })
}
eg.Wait()

如果您不知道有多少个输出，则此方法有效，但它不能保证有关输出顺序的任何信息 - 它可以是任何顺序。如果你想将其排序，你可以在所有 goroutine 完成后进行某种排序。