如何使用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 启动时和实际开始运行之间发生变化.

更新：从 Go 1.22 开始，这不再是问题，因为循环变量已更改为具有每次迭代范围。您可以在发行说明中查看详细信息。

---

其次，您同时从不同的 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 完成后进行某种排序。

Answer 2

运行某些 Go 例程时，无法保证顺序。因此，虽然可以预期元素

1

、

2

、

3

，但您不应该对顺序做出任何假设。

无论如何，看起来第一个

eg.Go()

调用是在 for

for

循环实际上到达其第三个元素时发生的。这就是为什么您只能获得

3

，并且只能在第三个位置（其中

i=2

）进行索引访问。

如果你像这样复制你的值，问题就得到了一定程度的解决：

for i, n := range input {
    nc, ic := n, i
    eg.Go(func() (err error) {
        output1 = append(output1, nc+1)
        output2[ic] = nc + 1
        return nil
    })
}

也就是说，结果看起来像这样

with append [3 2 1]
with make [1 2 3]

对我来说，所以订单仍然不是我们预期的。 不过，我不是

errgroup

包方面的专家，所以也许其他人可以分享有关执行顺序的更多信息。