优雅地关闭通道而不是在关闭的通道上发送

Question

我是Golang并发的新手，一直在努力理解下面提到的这段代码。

我目睹了一些我无法解释为什么会发生的事情：

1. for i <= 100000 {在 main 函数中 使用小于等于 100000 的 i 时，它有时会为 nResults 和 countWrites 打印不同的值（在最后两个语句中）fmt.Printf("number of result writes %d\n", nResults) fmt.Printf("Number of job writes %d\n", jobWrites)

2. 当我使用超过 1000000 时，它给出panic: send on closed channel

我如何确保发送给作业的值不在关闭的通道上，并且稍后在结果中收到所有值后我们可以关闭通道而不会出现死锁？

package main

import (

    "fmt"

    "sync"

)

func worker(wg *sync.WaitGroup, id int, jobs <-chan int, results chan<- int, countWrites *int64) {

    defer wg.Done()

    for j := range jobs {

        *countWrites += 1

        go func(j int) {

            if j%2 == 0 {

                results <- j * 2

            } else {

                results <- j

            }

        }(j)

    }

}

func main() {

    wg := &sync.WaitGroup{}

    jobs := make(chan int)

    results := make(chan int)

    var i int = 1

    var jobWrites int64 = 0

    for i <= 10000000 {

        go func(j int) {

            if j%2 == 0 {

                i += 99

                j += 99

            }

            jobWrites += 1

            jobs <- j

        }(i)

        i += 1

    }

    var nResults int64 = 0

    for w := 1; w < 1000; w++ {

        wg.Add(1)

        go worker(wg, w, jobs, results, &nResults)

    }

    close(jobs)

    wg.Wait()

    var sum int32 = 0

    var count int64 = 0

    for r := range results {

        count += 1

        sum += int32(r)

        if count == nResults {

            close(results)

        }

    }

    fmt.Println(sum)

    fmt.Printf("number of result writes %d\n", nResults)

    fmt.Printf("Number of job writes %d\n", jobWrites)

}

Answer 1

您的代码中有很多问题。在关闭的频道上发送使用 Go 通道的一个一般原则是不要从接收端关闭通道，如果通道有多个并发发送者则不要关闭通道您的解决方案很简单：不要有多个并发发件人，然后您可以从发件人端关闭频道。与其为添加到通道的每个作业启动数百万个单独的 goroutine，不如运行一个执行整个循环的 goroutine 以将所有作业添加到通道。并在循环后关闭通道。工作人员将尽可能快地使用通道。通过修改多个 goroutine 中的共享变量来进行数据竞争您在不采取特殊步骤的情况下修改两个共享变量：nResults，您将其传递给countWrites *int64工作人员中的。ijobs在写入作业通道的循环中：您从多个 goroutines 向它添加 99，这使得它无法预测您实际向通道写入了多少值要解决 1，有很多选项，包括使用sync.Mutex. 但是，由于您只是添加它，所以最简单的解决方案是使用atomic.AddInt64(countWrites, 1)而不是*countWrites += 1要解决 2，不要在每次写入通道时使用一个 goroutine，而是在整个循环中使用一个 goroutine（见上文）