Goroutines 睡着了

Question

我收到致命错误：所有 goroutine 都在睡觉 - 死锁！，即使我正在使用缓冲通道并获得结果。但是在得到所有结果后，我得到了错误。package mainimport (    "fmt"    "time")func main() {    StartJobExample()}var MAX_NUM = 1000func StartJobExample(){    t1 := time.Now()    jobs := make(chan int,MAX_NUM)    result:=make( chan int,MAX_NUM)    go worker(jobs,result)    writeNums(jobs)    for j := range result{        fmt.Println("number read is ",j)    }    close(result)   t2 := time.Now()   fmt.Println("Time taken in operation ",t2.Sub(t1).Seconds())}func  worker(jobs <-chan int ,result chan<- int){         for i:= range jobs{                result<-addTwo(i)         }}func writeNums(jobs chan<- int){    for i:=0;i<MAX_NUM;i++{        fmt.Println("adding job ",i)        jobs<-i    }    close(jobs)}func addTwo(i int)int{    return i+2}

Answer 1

您的代码中有几个错误，所以让我们一一检查。

首先，在您的情况下，您实际上并不需要缓冲通道，因为它们目前正在帮助掩盖您的真正问题。

其次，该writeNums函数应该在单独的 goroutine 中执行。现在，您在同一个 goroutine（ goroutine）中写入jobs通道并从通道读取，在您的情况下，这必然会产生死锁。使用缓冲通道掩盖了这个问题。因此，只需执行以下操作：resultmain

go writeNums(jobs)

第三，你当前的代码永远不会执行这个：

for j := range result{

        fmt.Println("number read is ",j)

    }

    close(result)

这就是在您的示例中产生死锁的原因。result通道永远不会关闭，因为在循环close(result)之后执行。for你应该把它close(result)放在worker函数中，在for循环之后。

func  worker(jobs <-chan int ,result chan<- int){

         for i:= range jobs{

                result<-addTwo(i)

         }

         close(result)

}

这样，result通道将在jobs通道关闭后立即关闭，并且for循环退出。

这是一个使用无缓冲通道的工作示例。您可以修改它以包含缓冲区，但没有它也可以工作。

编辑：

多工人解决方案

添加更多workergoroutine 的主要问题是关闭result. 由于现在有更多的workergoroutines 在运行，所有的workergoroutines 都需要在通道关闭result之前完成将值放入通道。result

这可以通过类似的同步机制来完成sync.WaitGroup。以下是更改：

func StartJobExample(){

    t1 := time.Now()

    numberOfWorkers := 5

    jobs := make(chan int)

    result:=make( chan int)

    var wg sync.WaitGroup

    //initialize a goroutine for syncing completion for all workers

    go func() {

      wg.Wait()

      close(result)

    }()

    

    //initialize all workers

    for i :=0; i < numberOfWorkers; i++ {

      wg.Add(1)

      go worker(&wg, jobs, result, i)

    }

// rest of the StartExample function is unchanged

// ...

}

func  worker(wg *sync.WaitGroup, jobs <-chan int ,result chan<- int, num int){

    for i:= range jobs{

           result<-addTwo(i)

    }

    wg.Done()

    fmt.Println("closing worker ", num)

}

在这个例子中，你使用sync.WaitGroup来同步所有的workergoroutine。您首先添加一个单独的 goroutine 来等待所有workergoroutine 完成（wg.Wait()）并关闭result通道。

其次，在你创建每个workergoroutine 之前，你需要先做wg.Add(1)。

第三，修改worker函数，使其wg.Done()在完成从通道读取jobs和写入result通道后执行。

当最后一个workergoroutine 执行wg.Done()时，wg.Wait()将在其 goroutine 中解除阻塞并执行close(result)。这是它如何工作的完整示例。