父通道消息未传递到子goroutine

对于一个Golang开发者来说，牢固扎实的基础是十分重要的，golang学习网就来带大家一点点的掌握基础知识点。今天本篇文章带大家了解《父通道消息未传递到子goroutine》，主要介绍了，希望对大家的知识积累有所帮助，快点收藏起来吧，否则需要时就找不到了！

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

client --> api --> handler --> worker1
                          |--> worker2
                          |--> worker3

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

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

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

注意： loglistenmgm 是我需要在工作人员中接收其消息的通道，其缓冲区的大小与工作人员的数量相同，但是如果我删除该缓冲区，则在写入 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 通道时，工作人员正在等待接收 stream.Read 上的新消息，并且无法从 loglistenmgmz 读取qbendczqb 频道。我通过在处理程序关闭后几分钟向 stream.Read 发送一些消息来确认这一点，然后工作人员能够从通道中读取并退出。

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

到这里，我们也就讲完了《父通道消息未传递到子goroutine》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！