使用goroutines处理值并将结果收集到切片中

Question

我最近正在探索Go，goroutines如何工作使我感到困惑。

我尝试使用goroutines将之前编写的代码移植到Go中，但出现fatal error: all goroutines are asleep - deadlock!错误。

我正在尝试使用goroutines处理列表中的项目，然后将处理后的值收集到新列表中。但是我在“聚会”部分遇到了问题。

码：

sampleChan := make(chan sample)
var wg sync.WaitGroup

// Read from contents list
for i, line := range contents {
    wg.Add(1)
    // Process each item with a goroutine and send output to sampleChan
    go newSample(line, *replicatePtr, *timePtr, sampleChan, &wg)
}
wg.Wait()

// Read from sampleChan and put into a slice
var sampleList []sample
for s := range sampleChan {
    sampleList = append(sampleList, s)
}
close(sampleChan)

从goroutine收集结果的正确方法是什么？

我知道切片不是线程安全的，所以我不能让每个goroutine仅追加到切片中。

Answer 1

有两个问题

1. 使用无缓冲通道：无缓冲通道阻塞接收器，直到通道上的数据可用，发送器直到接收器可用。这导致了错误
2. 在 range 之前不关闭通道：由于您从不关闭 ch 通道，因此范围循环将永远不会完成。

您必须在范围之前使用buffered频道和close频道

代码

package main

import (
    "fmt"
    "sync"
)

func double(line int, ch chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    ch <- line * 2

}

func main() {
    contents := []int{1, 2, 3, 4, 5}
    sampleChan := make(chan int,len(contents))
    var wg sync.WaitGroup
    // Read from contents list
    for _, line := range contents {
        wg.Add(1)
        go double(line, sampleChan, &wg)
    }
    wg.Wait()
    close(sampleChan)
    // Read from sampleChan and put into a slice
    var sampleList []int

    for s := range sampleChan {
        sampleList = append(sampleList, s)
    }

    fmt.Println(sampleList)
}

播放链接：https : //play.golang.org/p/k03vt3hd3P

编辑：有更好的表现另一种方法是运行producer和consumer在同时

修改代码

package main

import (
    "fmt"
    "sync"
)

func doubleLines(lines []int, wg *sync.WaitGroup, sampleChan chan int) {
    defer wg.Done()

    defer close(sampleChan)
    var w sync.WaitGroup
    for _, line := range lines {
        w.Add(1)
        go double(&w, line, sampleChan)
    }
    w.Wait()
}

func double(wg *sync.WaitGroup, line int, ch chan int) {
    defer wg.Done()
    ch <- line * 2
}

func collectResult(wg *sync.WaitGroup, channel chan int, sampleList *[]int) {
    defer wg.Done()
    for s := range channel {
        *sampleList = append(*sampleList, s)
    }

}

func main() {
    contents := []int{0,1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19}
    sampleChan := make(chan int, 1)
    var sampleList []int

    var wg sync.WaitGroup

    wg.Add(1)
    go doubleLines(contents, &wg, sampleChan)
    wg.Add(1)
    go collectResult(&wg, sampleChan, &sampleList)
    wg.Wait()
    fmt.Println(sampleList)
}

播放链接：https : //play.golang.org/p/VAe7Qll3iVM

Answer 2

您的代码几乎是正确的。存在两个问题：首先，您在收集结果之前等待所有工作程序完成，其次，for当通道关闭时，循环终止，但仅在for循环终止后才关闭通道。

您可以通过在工作程序完成后异步关闭通道来修复代码：

for i, line := range contents {
    wg.Add(1)
    // Process each item with a goroutine and send output to sampleChan
    go newSample(line, *replicatePtr, *timePtr, sampleChan, &wg)
}

go func() {
    wg.Wait()
    close(sampleChan)
}()

for s := range sampleChan {
  ..
}

作为样式的注释（并遵循https://github.com/golang/go/wiki/CodeReviewComments#synchronous-functions），最好newSample是一个简单的同步函数，它不占用waitgroup和channel ，并简单地产生其结果。然后，工作程序代码将如下所示：

for i, line := range contents {
    wg.Add(1)
    go func(line string) {
        defer wg.Done()
        sampleChan <- newSample(line, *replicatePtr, *timePtr)
    }(line)
}

这使您的并发原语保持在一起，除了简化newSample并使其易于测试之外，还使您可以查看并发情况，并直观地查看wg.Done()始终被调用的内容。而且，如果您希望将代码重构为例如使用固定数量的工作程序，则所做的更改将全部在本地进行。