Golang并发：如何从不同的goroutine附加到同一个切片

Question

我有并发 goroutines 想要将一个（指向一个）结构​​的（指针）附加到同一个切片。 你如何在 Go 中编写它以使其并发安全？

这将是我的并发不安全代码，使用等待组：

var wg sync.WaitGroup
MySlice = make([]*MyStruct)
for _, param := range params {
    wg.Add(1)
    go func(param string) {
        defer wg.Done()
        OneOfMyStructs := getMyStruct(param)
        MySlice = append(MySlice, &OneOfMyStructs)
    }(param)
}
wg.Wait()

我猜你需要使用 go 通道来保证并发安全。任何人都可以举个例子吗？

Answer 1

频道是解决此问题的最佳方式。这是一个可以在 go playground 上运行的示例。

package main

import "fmt"
import "sync"
import "runtime"

type T int

func main() {
    var slice []T
    var wg sync.WaitGroup

    queue := make(chan T, 1)

    // Create our data and send it into the queue.
    wg.Add(100)
    for i := 0; i < 100; i++ {
        go func(i int) {
            defer wg.Done()

            // Do stuff.
            runtime.Gosched()

            queue <- T(i)
        }(i)
    }

    // Poll the queue for data and append it to the slice.
    // Since this happens synchronously and in the same
    // goroutine/thread, this can be considered safe.
    go func() {
        defer wg.Done()
        for t := range queue {
            slice = append(slice, t)
        }
    }()

    // Wait for everything to finish.
    wg.Wait()

    fmt.Println(slice)
}

注意：runtime.Gosched() 调用存在是因为这些 goroutine 不会屈服于调度程序。如果我们没有明确地做一些事情来触发所述调度程序，这将导致死锁。另一种选择可能是执行一些 I/O（例如：打印到标准输出）。但我发现runtime.Gosched() 的意图更简单、更清晰。

Answer 2

使用sync.Mutex 保护MySlice = append(MySlice, &OneOfMyStructs) 没有任何问题。但是当然你可以有一个缓冲区大小为len(params)的结果通道，所有goroutines都会发送他们的答案，一旦你的工作完成，你就会从这个结果通道中收集。

如果您的params 具有固定大小：

MySlice = make([]*MyStruct, len(params))
for i, param := range params {
    wg.Add(1)
    go func(i int, param string) {
         defer wg.Done()
         OneOfMyStructs := getMyStruct(param)
         MySlice[i] = &OneOfMyStructs
     }(i, param)
}

由于所有的 goroutine 都写入不同的内存，这并不奇怪。

Answer 3

@jimt 发布的答案不太正确，因为它错过了通道中发送的最后一个值，并且最后一个 defer wg.Done() 从未被调用。下面的 sn-p 有更正。

https://play.golang.org/p/7N4sxD-Bai

package main

import "fmt"
import "sync"

type T int

func main() {
    var slice []T
    var wg sync.WaitGroup

    queue := make(chan T, 1)

    // Create our data and send it into the queue.
    wg.Add(100)
    for i := 0; i < 100; i++ {
        go func(i int) {
            // defer wg.Done()  <- will result in the last int to be missed in the receiving channel
            queue <- T(i)
        }(i)
    }

    go func() {
        // defer wg.Done() <- Never gets called since the 100 `Done()` calls are made above, resulting in the `Wait()` to continue on before this is executed
        for t := range queue {
            slice = append(slice, t)
            wg.Done()   // ** move the `Done()` call here
        }
    }()

    wg.Wait()

    // now prints off all 100 int values
    fmt.Println(slice)
}