避免重复输出的 Go 并发工作池

今天golang学习网给大家带来了《避免重复输出的 Go 并发工作池》，其中涉及到的知识点包括等等，无论你是小白还是老手，都适合看一看哦~有好的建议也欢迎大家在评论留言，若是看完有所收获，也希望大家能多多点赞支持呀！一起加油学习~

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

该程序按以下流程工作：

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

如果我在此处取消计算 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)
}

我一直在寻找为什么该程序没有显示一致的结果，但我还无法弄清楚。我如何更改程序才能产生相同的结果？

正确答案

您必须按以下方式重写 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) 之前完成。

countword 函数始终返回 count == 1 的结果。

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

func countword(word string, tempmap map[string]int) result {
    count := tempmap[word] + 1
    tempmap[word] = count
    return result{word, count}
}

但是保持这个想法！ comcomputitatal假设cbcountzqbentzqbendcbendczqb的结果始终发送result中的工人始终发送result {word，1} ult {word，1} 直接来自 readtext。代码如下：

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。

今天关于《避免重复输出的 Go 并发工作池》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！