子goroutine不从父通道接收消息

Question

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

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

我使用不同的通道在所有这些goroutine之间进行通信，工人可以写到聚合通道，处理程序（父goroutine）可以接收它并将其发送回客户机，所以一切都正常工作，几乎是：）

问题是我的工人没有收到由通道上的处理程序发送的消息，工人的goroutines从未停止，但我找不到他们为什么没有收到消息。

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
        }
    }

}

Answer 1

由于https://stackoverflow.com/users/32880/jimb指出问题在于stream.read并阻止了select，所以我找到了问题并解决了它。

stream.read是频率较低的IO.readcloser，因此当我将true事件发送到loglistenmgm通道时，工作人员正在等待在stream.read上接收新消息，无法从loglistenmgm通道读取。我通过在处理程序关闭几分钟后向stream.read发送一些消息来确认这一点，然后工作人员可以从通道读取并退出。

我通过更改程序解决了这个问题，我没有在worker中创建流，而是在处理程序中创建它们并将它们传递给worker，当处理程序完成后，我关闭所有流，这将触发numbytes==0并关闭worker。