从同一通道读取的多个 Go 例程

Question

嗨，我遇到了（某种）控制通道的问题。

我的程序精华：

我不知道在运行时我会运行多少个go例程
我需要在设定的时间重新启动这些 go 例程，但是，它们也可能会出错（然后重新启动），因此它们的时间将无法预测。
这些 go 例程会将消息放入单个通道。

所以我所做的是创建一个简单的随机消息生成器来将消息放入频道。当计时器启动时（测试的随机持续时间）我将一条消息放到 控制通道 上，这是一个 结构有效负载 ，所以我知道有关闭信号，这是例行程序；实际上，在再次开始 go 例程之前，我会做一些其他我需要做的事情。

我的问题是：

我在 reflect.Select 循环中收到控制消息
我不（或无法）在我的 randmsgs() 循环中接收它

因此我无法停止我的 randmsgs() go 例程。

我认为我理解多个 go 例程可以从单个通道读取是正确的，因此我认为我误解了 reflect.Select案例如何适应所有这些。

我的代码：

package main

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

type testing struct {
    control bool
    market  string
}

func main() {
    rand.Seed(time.Now().UnixNano())
    // explicitly define chanids for tests.
    var chanids []string = []string{"GR I", "GR II", "GR III", "GR IV"}
    stream := make(chan string)
    control := make([]chan testing, len(chanids))
    reflectCases := make([]reflect.SelectCase, len(chanids)+1)

    // MAKE REFLECT SELECTS FOR 4 CONTROL CHANS AND 1 DATA CHANNEL
    for i := range chanids {
        control[i] = make(chan testing)
        reflectCases[i] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(control[i])}
    }
    reflectCases[4] = reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(stream)}
    
    // START GO ROUTINES
    for i, val := range chanids {
        runningFunc(control[i], val, stream, 1+rand.Intn(30-1))
    }

    // READ DATA
    for {
        o, recieved, ok := reflect.Select(reflectCases)
        if !ok {
            fmt.Println("You really buggered this one up...")
        }
        ty, err := recieved.Interface().(testing)
        if err == true {
            fmt.Printf("Read from: %v, and recieved close signal from: %s\n", o, ty.market)
            // close control & stream here.
        } else {
            ty := recieved.Interface().(string)
            fmt.Printf("Read from: %v, and recieved value from: %s\n", o, ty)
        }

    }
}

// THE GO ROUTINES - TIMER AND RANDMSGS
func runningFunc(q chan testing, chanid string, stream chan string, dur int) {
    go timer(q, dur, chanid)
    go randmsgs(q, chanid, stream)
}

func timer(q chan testing, t int, message string) {
    for t > 0 {
        time.Sleep(time.Second)
        t--
    }
    q <- testing{true, message}
}

func randmsgs(q chan testing, chanid string, stream chan string) {

    for {
        select {
        case <-q:
            fmt.Println("Just sitting by the mailbox. :(")
            return
        default:
            secondsToWait := 1 + rand.Intn(5-1)
            time.Sleep(time.Second * time.Duration(secondsToWait))
            stream <- fmt.Sprintf("%s: %d", chanid, secondsToWait)
        }
    }
}

我为文字墙道歉，但我完全没有想法:(!

K/Regards, C.

Answer 1

您下半场的频道 q 与上半场的 control[0...3] 相同。

你的 reflect.Select 即你运行也从所有这些频道读取，没有延迟。

我认为问题归结为您的 reflect.Select 只是运行太快了，立即“窃取”了所有通道输出。这就是 randmsgs 永远无法阅读邮件的原因。

您会注意到，如果您从 randmsgs 中删除默认大小写，该函数能够（可能）读取来自 q 的一些消息。

        select {
        case <-q:
            fmt.Println("Just sitting by the mailbox. :(")
            return
        }

这是因为现在运行没有延迟，它一直在等待q的消息，因此有机会在比赛中击败reflect.Select。

如果您在多个 goroutine 中从同一个通道读取，那么传递的数据将简单地转到首先读取它的 goroutine。

这个程序似乎只是一个实验/学习经验，但我会提供一些可能有帮助的批评。

同样，如果两个 goroutine 正在执行不同的任务，通常您不会有多个 goroutines 从同一通道读取数据。您正在创建一个关于哪个 goroutine 首先获取数据的不确定性竞赛。
其次，这是一个常见的初学者反模式select，你应该避免：

for {
    select {
    case v := <-myChan:
        doSomething(v)
    default:
        // Oh no, there wasn't anything! Guess we have to wait and try again.
        time.Sleep(time.Second)
}

此代码是多余的，因为 select 已经以这样一种方式运行：如果没有案例最初准备就绪，它将等到任何案例准备就绪，然后继续处理该案例。 default: sleep 有效地使您的 select 循环变慢，但实际等待通道的时间却减少了（因为 99.999...% 的时间花在 time.Sleep 上）。

从同一通道读取的多个 Go 例程

Multiple Go routines reading from the same channel

go

goroutine