不产生相同的输出 Concurrency Go worker pool

Question

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

该程序按以下流程运行：

读取文本文件（readText 函数）
readText 函数将每个单词发送到 word 通道
每个 goroutine 执行 countWord 计算映射中单词的函数
每个goroutine returns一个map和worker函数将struct的Result值传递给resultC通道
测试函数根据来自 resultC 通道
打印第 5 步创建的地图

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

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

程序在没有showing/counting所有单词

的情况下终止

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 函数总是 returns 计数 == 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}，我们可以将工人从图片中删除通过直接从 readText 发送 Result{word, 1}。这是代码：

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.

通道操作的开销可能会抵消运行 computeTotal 和 readText 在单独的 goroutine 中的任何好处。这是合并到一个 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 函数让我认为您的目标是计算每个 worker 中的单词数并将结果合并为总数。这是相关代码：

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) 之前完成。

不产生相同的输出 Concurrency Go worker pool

Does not produce the same output Concurrency Go worker pool

concurrency

go