不会产生相同的输出 Concurrency Go 工作池

Question

我正在编写一个程序，它同时从文本文件中逐字读取，以使用通道和工作池模式计算出现次数

程序按以下流程运行：

1. 读取文本文件（readText 函数）
2. readText 函数将每个单词发送到word 频道
3. 每个 goroutine 都执行 countWord 函数来计算地图中的单词
4. 每个goroutine返回一个map，worker函数将struct的Result值传递给resultC通道
5. 测试函数根据来自resultC 频道的结果值创建地图
6. 打印从第 5 步创建的地图

程序可以运行，但是当我尝试输入fmt.Println(0)时，看到的过程如下图

func computeTotal() {
    i := 0
    for e := range resultC {
        total[e.word] += e.count
        i += 1
        fmt.Println(i)
    }
}

程序终止而不显示/计算所有单词

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 all goroutines finished 16 17 18 map[but:1 cat's:1 crouched:1 fur:1 he:2 imperturbable:1 it:1 pointed:1 sat:1 snow:1 stiffly:1 the:1 was:2 with:1] total words: 27 38 ... 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 Time taken for reading the book 5.8145ms

如果我在这里取消对compute Total函数语句中的fmt.println()的注释，程序会正确显示结果，输出如下所示

all goroutines finished
map[a:83 about:4 above:2 absolute:1 accepted:1 across:1 affection:1 after:1 again:5  wonder:2 wood:5 wooded:1 woody:1 work:1 worked:2 world:4 would:11 wrapped:1 wrong:1 yellow:2 yielded:1 yielding:1 counts continues ......]
total words:  856
Time taken for reading the book 5.9924ms

这是我对 readtext 的实现

//ensure close words at the right timing
func readText() {

    file, err := os.Open(FILENAME)
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()
    scanner := bufio.NewScanner(file)
    scanner.Split(bufio.ScanWords)

    for scanner.Scan() {
        word := strings.ToLower(scanner.Text())
        words <- strings.Trim(word, ".,:;")

    }
    //time.Sleep(1 * time.Second)
    close(words)
}

这是我使用工作池实现的计数字

//call countWord func,
func workerPool() {
    var wg sync.WaitGroup
    for i := 1; i <= NUMOFWORKER; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    fmt.Println("all goroutines finished")
    close(resultC)
}

func worker(wg *sync.WaitGroup) {
    var tempMap = make(map[string]int)
    for w := range words {
        resultC <- countWord(w, tempMap) //retuns Result value
    }
    wg.Done()

}

//creates a map each word
func countWord(word string, tempMap map[string]int) Result {
    _, ok := tempMap[word]
    if ok {
        tempMap[word]++
        return Result{word, tempMap[word] + 1}

    }
    return Result{word, 1}

}

最后，这是主要功能

const FILENAME = "cat.txt"
const BUFFERSIZE = 3000
const NUMOFWORKER = 5

var words = make(chan string, BUFFERSIZE) //job
var resultC = make(chan Result, BUFFERSIZE)

var total = map[string]int{}

type Result struct {
    word  string
    count int
}

func main() {
    startTime := time.Now()
    go readText()
    go computeTotal()
    workerPool() //blocking
    fmt.Println(total)
    endTime := time.Now()
    timeTaken := endTime.Sub(startTime)
    fmt.Println("total words: ", len(total))
    fmt.Println("Time taken for reading the book", timeTaken)
}

我一直在寻找程序为什么没有显示一致的结果，但我还没有弄明白。如何更改程序以使其产生相同的结果？

Answer 1

countWord 函数总是返回 count == 1 的结果。

这是增加计数的函数的一个版本：

func countWord(word string, tempMap map[string]int) Result {
    count := tempMap[word] + 1
    tempMap[word] = count
    return Result{word, count}
}

但请保持这种想法！ computeTotal 函数假设结果 count 为 1。鉴于问题中的工人总是按照 computeTotal 的预期发送 Result{word, 1}，我们可以通过直接从 @ 发送 Result{word, 1} 将工人从图片中删除987654332@。代码如下：

func computeTotal() {
    i := 0
    for e := range resultC {
        total[e.word] += e.count
        i += 1
        fmt.Println(i)
    }
}

func readText() {
    file, err := os.Open(FILENAME)
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()
    scanner := bufio.NewScanner(file)
    scanner.Split(bufio.ScanWords)

    for scanner.Scan() {
        word := strings.ToLower(scanner.Text())
        resultC <- Result{strings.Trim(word, ".,:;"), 1}
    }
    close(resultC)
}

main() {
    ...
    go readText()
    computeTotal()
    fmt.Println(total)
    ...
}

Run it on the playground.

通道操作的开销可能抵消了在单独的 goroutine 中运行 computeTotal 和 readText 的任何好处。这是合并到单个 goroutine 中的代码：

func main() {
    file, err := os.Open(FILENAME)
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()
    scanner := bufio.NewScanner(file)
    scanner.Split(bufio.ScanWords)

    var total = map[string]int{}
    for scanner.Scan() {
        word := strings.ToLower(strings.Trim(scanner.Text(), ".,:;"))
        total[word]++
    }
    fmt.Println(total)
}

Run it on the playground.

问题中的countWord 函数让我认为您的目标是计算每个工人中的单词并将结果合并为一个总数。这是代码：

func computeTotal() {
    for i := 1; i <= NUMOFWORKER; i++ {
        m := <-resultC
        for word, count := range m {
            total[word] += count
        }
    }
}

func workerPool() {
    for i := 1; i <= NUMOFWORKER; i++ {
        go worker()
    }
}

func worker() {
    var tempMap = make(map[string]int)
    for w := range words {
        tempMap[w]++
    }
    resultC <- tempMap
}

...
var resultC = make(chan map[string]int)
...

func main() {
    ...
    go readText()
    workerPool()
    computeTotal()
    ...
}

Run it on the playground.

Answer 2

你必须用以下方式重写你的computeTotal函数：

func computeTotal(done chan struct{}) {
    defer close(done)
    i := 0
    for e := range resultC {
        total[e.word] += e.count
        i += 1
        fmt.Println(i)
    }
}

func main() {

   computeTotalDone := make(chan struct{})
   go computeTotal(computeTotalDone)
   ...
   workerPool() //blocking
   <-computeTotalDone
   fmt.Println(total)
}

添加fmt.Println 导致无效结果的原因是您的实现存在竞争条件。由于在主函数 fmt.Println(total) 和 computeTotal 函数中打印总结果并行运行，因此无法保证 computeTotal 在调用 fmt.Println(total) 之前处理所有消息。如果没有fmt.Println，computeTotal 函数在您的计算机上的速度足以产生正确的结果。

建议的解决方案确保computeTotal 在调用fmt.Println(total) 之前完成。