子 goroutine 没有收到来自父频道的消息

如何解决子 goroutine 没有收到来自父频道的消息

我有一个服务器，它在收到请求时需要使用 goroutines 从不同的流中读取消息，将它们发送到父 goroutine，父 goroutine 聚合消息并将它们发送到客户端。所以它会是这样的：

Client --> API --> handler --> worker1
                          |--> worker2
                          |--> worker3

我使用不同的通道在所有这些 goroutine 之间进行通信，worker 可以写入聚合通道，处理程序（父 goroutine）可以接收它并将其发送回客户端，因此一切正常，几乎 :)

问题是我的工作人员没有收到通道上处理程序发送的消息，工作人员的 goroutine 永远不会停止，但我找不到他们收不到消息的原因。

由于代码量很大，我只放了代码的相关部分。

注意： loglistenmgm 是我需要在 worker 中接收其消息的通道，其缓冲区大小与 worker 数量相同，但如果我删除缓冲区，则在向其写入 true 后信息日志将永远不会打印出来（这意味着没有 goroutine 监听），我只会在日志中看到这个，日志挂在那里：

INFO[0010] Request.Context().Done triggered             
INFO[0010] **Sending true event to loglistenmgm: 0

我的处理程序：

func handler() {
        logStream := make(chan string)
        done := make(chan bool,5)
        loglistenmgm := make(chan bool,numberOfPods)
        errorStream := make(chan error)

        for _,pod := range pods {
            go getOnePodLogs(logStream,errorStream,loglistenmgm,done)
        }

        go func() {
            for {
                select {
                case <-c.Request.Context().Done():
                    log.Info("Request.Context().Done triggered")
                    for i := 0; i < numberOfPods; i++ {
                        log.Info("**Sending true event to loglistenmgm: " + strconv.Itoa(i))
                        loglistenmgm <- true
                        log.Info("**After sending true event to loglistenmgm") // This gets printed so the true has been sent to the channel
                    }
                    done <- true
                    return
                case err := <-errorStream:
                    c.JSON(http.StatusInternalServerError,map[string]string{
                        "message": "Internal Server Error: " + err.Error(),})
                    for i := 0; i < numberOfPods; i++ {
                        loglistenmgm <- true
                    }
                    done <- true
                    return
                }
            }
        }()

        isStreaming := c.Stream(func(w io.Writer) bool {
            for {
                select {
                case <-done:
                    c.SSEvent("end","end") // This also works properly and it can print the "stream closed" which happens after this
                    return false
                case msg := <-logStream:
                    c.Render(-1,sse.Event{
                        Event: "message",Data:  msg,})
                    return true
                }
            }
        })
        if !isStreaming {
            log.Info("stream closed")
        }
}

我的员工：

func getOnePodLogs(logStream chan string,errorStream chan error,loglistenmgm chan bool,done chan bool) {
    stream,err := podLogRequest.Stream()
    defer stream.Close()

    if err != nil {
        log.Error(err.Error())
        errorStream <- err
        return
    }

    for {
        select {
        case <-loglistenmgm:
            log.Info(pod + "stop listenning to logs") // this log line never get triggered
            return
        default:
            buf := make([]byte,1000)
            numBytes,err := stream.Read(buf)
            if numBytes == 0 {
                log.Info(pod + ": numBytes == 0 --> End of log")
                done <- true
                return
            }
            if err == io.EOF {
                log.Info("io.EOF")
                return
            }
            if err != nil {
                log.Error("Error getting stream.Read(buf)")
                log.Error(err)
                return
            }
            message := string(buf[:numBytes])
            logStream <- message // This works and I can receive the message on handler and it can pass it to the client
        }
    }

}

解决方法

感谢 https://stackoverflow.com/users/32880/jimb 指出问题是 stream.Read 并且阻止了 select 我发现了问题并解决了它。

stream.Read 是一个频率较低的 io.ReadCloser，因此当我将 true 事件发送到 loglistenmgm 频道时，工作人员正在等待在 {{ 1}} 且无法从 stream.Read 通道读取。我通过在处理程序关闭后的 loglistenmgm 分钟发送一些消息来确认这一点，然后工作人员能够从通道读取并退出。

我通过更改程序解决了这个问题，而不是在工作程序中创建 stream.Read，而是在处理程序中创建它们并将它们传递给工作程序，当我完成处理程序时，我关闭所有流，这会触发 stream 并关闭工作线程。