go - 加载带有和不带有 go-routines 的 map

Question

这是我遇到的一个有趣的情况。在使用 go-routines 进行一些数据操作之后，我需要从文件中读取，并根据我们发现的内容填充 map 。这是简化的问题陈述和示例:

通过运行 gen_data.sh 生成所需的数据

#!/bin/bash 

rm some.dat || : 
for i in `seq 1 10000`; do 
    echo "$i `date` tx: $RANDOM rx:$RANDOM" >> some.dat
done

如果我在 some.dat 中阅读这些行变成 map[int]string没有使用 loadtoDict.go 的 go-routines ，它保持对齐。 (因为第一个和第二个词是一样的，见下面的o/p。)

在现实生活中，我确实需要在将线条加载到 map 之前对其进行处理(昂贵)，使用 go-routines 加快了我的字典创建速度，这是解决实际问题的重要要求。

loadtoDict.go

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
)

var (
    fileName = "some.dat"
)

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

func main() {
    ourDict := make(map[int]string)
    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()

    fscanner := bufio.NewScanner(f)

    indexPos := 1

    for fscanner.Scan() {
        text := fscanner.Text()
        //fmt.Println("text", text)
        ourDict[indexPos] = text
        indexPos++

    }

    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

运行:

$ ./loadtoDict
...
8676: 8676 Mon Dec 23 15:52:24 PST 2019 tx: 17718 rx:1133
2234: 2234 Mon Dec 23 15:52:20 PST 2019 tx: 13170 rx:15962
3436: 3436 Mon Dec 23 15:52:21 PST 2019 tx: 17519 rx:5419
6177: 6177 Mon Dec 23 15:52:23 PST 2019 tx: 5731 rx:5449

注意第一个和第二个词是如何“对齐”的。但是，如果我使用 go-routines 加载我的地​​图，这会出错:

async_loadtoDict.go

package main

import (
    "bufio"
    "fmt"
    "log"
    "os"
    "sync"
)

var (
    fileName = "some.dat"
    mu       = &sync.RWMutex{}
    MAX = 9000
)

func checkerr(err error) {
    if err != nil {
        fmt.Println(err)
        log.Fatal(err)
    }
}

func main() {
    ourDict := make(map[int]string)
    f, err := os.Open(fileName)
    checkerr(err)
    defer f.Close()

    fscanner := bufio.NewScanner(f)

    indexPos := 1
    var wg sync.WaitGroup
    sem := make(chan int, MAX)
    defer close(sem)

    for fscanner.Scan() {
        text := fscanner.Text()
        wg.Add(1)
        sem <- 1
        go func() {
            mu.Lock()
            defer mu.Unlock()
            ourDict[indexPos] = text
            indexPos++
            <- sem
            wg.Done()
        }()

    }

    wg.Wait()

    for i, v := range ourDict {
        fmt.Printf("%d: %s\n", i, v)
    }

}

输出:

$ ./async_loadtoDict 
...
11: 22 Mon Dec 23 15:52:19 PST 2019 tx: 25688 rx:7602
5716: 6294 Mon Dec 23 15:52:23 PST 2019 tx: 28488 rx:3572
6133: 4303 Mon Dec 23 15:52:21 PST 2019 tx: 24286 rx:1565
7878: 9069 Mon Dec 23 15:52:25 PST 2019 tx: 16863 rx:24234
8398: 7308 Mon Dec 23 15:52:23 PST 2019 tx: 4321 rx:20642
9566: 3489 Mon Dec 23 15:52:21 PST 2019 tx: 14447 rx:12630
2085: 2372 Mon Dec 23 15:52:20 PST 2019 tx: 14375 rx:24151

尽管保护了摄取 ourDict[indexPos]与互斥锁。我希望我的 map 索引与摄取尝试保持一致。

谢谢!

Answer 1

您的信号量 sem不起作用，因为您对其进行了深度缓冲。

一般来说，这是为此类任务设置 map 的错误方法，因为读取文件会很慢。如果你有一个更复杂的任务——例如，读一行，想很多，设置一些东西——你会想要这个作为你的伪代码结构:

type workType struct {
    index int
    line  string
}

var wg sync.WaitGroup
wg.Add(nWorkers)
// I made this buffered originally but there's no real point, so
// fixing that in an edit
work := make(chan workType)
for i := 0; i < nWorkers; i++ {
    go readAndDoWork(work, &wg)
}

for i := 1; fscanner.Scan(); i++ {
    work <- workType{index: i, line: fscanner.Text()}
}
close(work)
wg.Wait()

... now your dictionary is ready ...

worker 这样做:

func readAndDoWork(ch chan workType, wg *sync.WorkGroup) {
    for item := range ch {
        ... do computation ...
        insertIntoDict(item.index, result)
    }
    wg.Done()
}

与 insertIntoDict抓取互斥锁(以保护映射从索引到结果)并写入字典。 (如果你愿意，你可以内联它。)

这里的想法是设置一定数量的工作人员——可能基于可用 CPU 的数量——每个工作人员抓取下一个工作项并处理它。主 goroutine 只是打包工作，然后关闭工作 channel ——这将导致所有工作人员看到输入结束——然后等待他们发​​出信号表明他们已经完成了计算。

(如果您愿意，您可以再创建一个 goroutine 来读取工作人员计算的结果并将它们放入映射中。这样您就不需要映射本身的互斥体。)