通过通道将值发送到多个 goroutine

Question

我想在通道中发送一个值以从主函数执行例程。发生的情况是哪个 goroutine 将首先从通道接收值。

package main

import (
    "fmt"
    "math/rand"
    //"runtime"
    "strconv"
    "time"
)

func main() {
    var ch chan int
    ch = make(chan int)
    ch <- 1
    receive(ch)
}

func receive(ch chan int){
    for i := 0; i < 4; i++ {
        // Create some threads
        go func(i int) {
            time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
            fmt.Println(<-ch)
        }(i)
    }
}

我当前的实现出错了。

致命错误：所有 goroutine 都处于休眠状态 - 死锁！

我怎么知道哪个 goroutine 将首先从通道接收值。其他 goroutine 会发生什么如果因为没有通道接收值而运行或抛出错误。因为其中一个已经收到了。

如果创建缓冲通道，我的代码可以工作。所以我不明白幕后发生了什么，这使它在创建如下所示的缓冲通道时起作用：

func main() {
    var ch chan int
    ch = make(chan int, 10)
    ch <- 1
    receive(ch)
}

如果我们看下面的代码。我可以看到我们可以直接通过通道发送值，不需要创建一个 goroutine 来通过一个通道向另一个 goroutine 发送一个值。

package main

import "fmt"

func main() {

    // We'll iterate over 2 values in the `queue` channel.
    queue := make(chan string, 2)
    queue <- "one"
    queue <- "two"
    close(queue)

    for elem := range queue {
        fmt.Println(elem)
    }
}

那么我的代码有什么问题。为什么会造成死锁。

Answer 1

如果您只需要启动多个工作人员并向其中任何个工作人员发送任务，那么您最好在向通道发送值之前运行工作人员，因为正如@mkopriva 上面所说，写入通道是一个阻塞操作。您始终必须有一个消费者，否则执行将冻结。

func main() {
    var ch chan int
    ch = make(chan int)

    receive(ch)

    ch <- 1
}

func receive(ch chan int) {
    for i := 0; i < 4; i++ {
        // Create some threads
        go func(i int) {
            time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
            fmt.Printf("Worker no %d is processing the value %d\n", i, <-ch)
        }(i)
    }
}

对于“哪个 goroutine 会收到它？”这个问题的简短回答- 随便。 :) 其中任何一个，你都不能确定。

但是我不知道什么是 time.Sleep(...) 在那里，保持原样。

Answer 2

一个无缓冲的通道（没有长度）阻塞，直到接收到值。这意味着写入通道的程序将在写入通道后停止，直到它被读取为止。如果这发生在主线程中，在您调用 receive 之前，它会导致死锁。

还有两个问题：您需要使用WaitGroup 来暂停完成直到完成，并且通道的行为类似于并发队列。特别是它有 push 和 pop 操作，这两个操作都是使用<- 执行的。例如：

//Push to channel, channel contains 1 unless other things were there
c <- 1
//Pop from channel, channel is empty
x := <-c

这是一个工作示例：

package main

import (
        "fmt"
        "math/rand"
        "sync"
        "time"
)

func main() {
        var ch chan int 
        ch = make(chan int)
        go func() {
                ch <- 1
                ch <- 1
                ch <- 1
                ch <- 1
        }() 
        receive(ch)
}

func receive(ch chan int) {
        wg := &sync.WaitGroup{}
        for i := 0; i < 4; i++ {
                // Create some threads
                wg.Add(1)
                go func(i int) {
                        time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
                        fmt.Println(<-ch)
                        wg.Done()
                }(i)
        }   
        wg.Wait()
        fmt.Println("done waiting")
}

Playground Link

如您所见，WaitGroup 也很简单。您在更高的范围内声明它。它本质上是一个花哨的计数器，具有三种主要方法。当你调用wg.Add(1)时计数器增加，当你调用wg.Done()时计数器减少，当你调用wg.Wait()时停止执行直到计数器达到0。