从通道读取的多个 goroutine 给出错误的数据计数答案

【问题标题】：Multiple goroutines reading from a channel gives wrong data count从通道读取的多个 goroutine 给出错误的数据计数
【发布时间】：2019-01-01 09:20:16
【问题描述】：

我正在开发一个读取 csv 文件并执行以下操作的程序：

完整代码在：Here

我的 CSV 文件位于： CSV file

问题是有时我得到正确的 A 和 B 计数，有时我得到错误的计数。

我认为我在 Goroutine 和通道通信方面做错了。

当我评论第二个协程时，我得到了第一个协程的正确结果。但是当我取消注释 2nd Goroutine 时，我得到的 Goroutine 1 和 2 的输出都不正确。

谁能解释一下我做错了什么？

另外，当我运行 -race main.go 时，结果显示了一个竞态条件。

func main() {
    input, err := os.Open("CSV.csv")
    if err != nil {
        fmt.Println("Error while opening CSV file.")
        return
    }
    defer input.Close()

    formattedStartDateRange,err := time.Parse(time.RFC3339, startDateRange)
    if err != nil {
        fmt.Println(err)
    }

    formattedendDateRange,err := time.Parse(time.RFC3339, endDateRange)
    if err != nil {
        fmt.Println(err)
    }

    reader := csv.NewReader(input)
    reader.FieldsPerRecord = -1
    files := make(map[string]chan []string)

    wg := &sync.WaitGroup{}

    var line []string
    for line, err = reader.Read(); err == nil; line, err = reader.Read() {
        ch, ok := files[line[0]]
        if ok {
            ch <- line
        } else {
            ch = make(chan []string, 8)
            ch <- line
            wg.Add(2) // Must wait for 2 calls to 'done' before moving on

            go func() {
                UserMapMutex.Lock()
                if (findNumberOfBuilds(formattedStartDateRange, formattedendDateRange, ch, wg)) {
                    totalBuildCount++
                }
                UserMapMutex.Unlock()
                wg.Done()
            }()

            go func() {
                UserMapMutex.Lock()
                countUserBuildFrequency(ch, wg)
                UserMapMutex.Unlock()
                wg.Done()
            }()


            files[line[0]] = ch
        }
    }


    if err.Error() != "EOF" {
        fmt.Println("Error while reading CSV file.")
        return
    }
    for _, ch := range files {
        close(ch)
    }
    wg.Wait()

    fmt.Println("Total Build executed from 1st November to 30th November =", totalBuildCount)
    fmt.Println("Total Build", userBuildFreq["5c00a8f685db9ec46dbc13d7"])
    fmt.Println("Done!")
}

【问题讨论】：

所有问题必须完整，无需点击链接。这意味着您必须在问题中包含您的代码和 CSV 文件的Minimal, Complete, Verifiable Example。提供更大版本的链接当然是受欢迎的，但关键是问题必须独立完成，以防将来链接失效。
这段代码很奇怪。对于每一行，您创建一个 goroutine，将一行写入通道，然后每个 goroutine 调用一个函数，该函数以某种未显示的方式使用该通道。这 a) 意味着函数调用可能会或可能不会使用刚刚插入的行 - 您可能应该只传递行而不是使用通道； b) 意味着如果其中一个函数从通道读取值，另一个函数将无法读取，因为接收次数是发送次数的两倍； c) 你运行两个完全锁定的 goroutine，所以你最好只使用一个。
正确！你提到的B选项就是我正在做的。我做了同样的事情，只使用了一个 goroutine。
但是有一个疑问，当我们有两个 goroutines 时，我们如何获得正确的结果？我知道如果一个 goroutines 从通道读取，那么其他将无法读取。有什么办法吗？

标签： go concurrency goroutine

【解决方案1】：

在这两种情况下，您的 wg.Done() 都会在您启动 goroutine 后立即被调用。这意味着您的 WaitGroup 没有等待 goroutine 完成。请记住，当您调用 goroutine 时，调用过程会继续进行。尝试将 wg.Done() 调用放入 goroutine，当它完成操作时。

go func(wg) {
    // do stuff
    wg.Done
}

或

go func(wg) {
    defer wg.Done
    // do stuff
}

【讨论】：

我已经按照您在下面的链接play.golang.org/p/8Mhjco_7RyM 中所说的那样更新了代码，但仍然没有成功。当我评论第二个协程时，我得到了第一个协程的正确结果。但是当我取消注释 2nd Goroutine 时，我得到的 Goroutine 1 和 2 的输出都不正确。
所以我认为是第二个 Goroutine 造成了混乱。
你有什么新发现吗？