【发布时间】：2021-06-26 12:22:45
【问题描述】：
我有一个服务器，它在收到请求后需要使用 goroutine 从不同的流中读取消息，将它们发送到父 goroutine，然后父 goroutine 聚合消息并将它们发送到客户端。所以它会是这样的：
Client --> API --> handler --> worker1
|--> worker2
|--> worker3
我使用不同的通道在所有这些 goroutine 之间进行通信，worker 可以写入聚合通道，处理程序（父 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
}
}
}
【问题讨论】：
-
从这里的例子中很难看出，但是当你处于意外状态时，我要做的第一件事是查看堆栈跟踪并查看所有内容被阻塞的位置。像这样使用一堆通道总是可能导致 goroutine 之间的同步错误。我建议你努力简化这个设计，避免尝试通过
done和loglistenmgm之类的通道在goroutine 之间“发送信号”，而更喜欢WaitGroup和Context之类的东西来控制goroutine。 -
我确切地知道问题出在哪里，但我不知道为什么会发生以及如何解决它。问题是处理程序（父 goroutine）将
true发送到loglistenmgm，但工作人员在loglistenmgm上没有收到任何消息。我会调查上下文。 -
我们知道消息不会因为渠道工作而丢失，所以某些东西必须处于您不期望的状态，这就是我试图描述的问题。最有可能的候选人是您在
stream.Read中被阻止，这意味着您将永远无法摆脱default的情况来检查loglistenmgm上的消息。 -
您对
stream.Read的看法可能是正确的，因为该流中消息的频率并不高。我尝试了sync.WaitGroup（我添加了所有必需的代码）但它不起作用，最后它也阻止了处理程序goroutine。有没有其他方法可以杀死这些工人？ -
sync.WaitGroup是等待 goroutines 返回，它不会解决你的问题。如何停止灌浆完全取决于podLogRequest.Stream()是什么。
标签： go concurrency goroutine