如何解决 websocket 上 write:broken pipeline 问题？

Question

我正在为我的程序制作一个进度条，并将其发送给客户。我使用的IPC是websocket。但是当我向客户端更新进度条时，我面临

write: broken pipe

。此错误发生在第一个请求之后。例如，在我的应用程序首次运行时，当我向服务器执行请求时，没有错误。但是，当我执行另一个请求时，就会出现此错误。

虽然有这个错误，但是进度条实际上是在工作的（进度条成功发送到客户端直到100%）

我使用 go Fiber websocket 作为我的服务器

// middleware
app.Use("/ws", func(c *fiber.Ctx) error {
  if websocket.IsWebSocketUpgrade(c) {
    c.Locals("allowed", true)
      return c.Next()
    }

    return fiber.ErrUpgradeRequired
})

// handler
func (s *downloaderService) progressBar(c *websocket.Conn) {
    channel := api.CreateChannel(c.Params("client"))

    for data := range channel.Subscribe() {
        progressBar := data.(downloader.Progressbar)

        payload, err := json.Marshal(progressBar)
        if err != nil {
            log.Println("Error marshalling data:", err)
            return
        }

        if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
            log.Println("Error sending progress data:", err)
            return
        }
    }
}

在客户端，我使用的是 gorilla websocket

interrupt := make(chan os.Signal, 1)
signal.Notify(interrupt, []os.Signal{syscall.SIGINT, syscall.SIGKILL, syscall.SIGTERM, syscall.SIGSTOP, os.Interrupt}...)

ctx, cancel := context.WithCancel(context.Background())

conn, res, err := websocket.DefaultDialer.DialContext(ctx, ws, nil)
if err != nil {
    log.Fatalf("Error dialing websocket: %v. Status courlde %d", err, res.StatusCode)
    return
}
defer conn.Close()

progressBar := progressbar()

go func() {
    for {
        _, message, err := conn.ReadMessage()
        if err != nil {
            log.Println("Error reading message:", err)
            break
        }

        var progress progress
        if err := json.Unmarshal(message, &progress); err != nil {
            log.Println("Error unmarshalling message:", err)
            break
        }

        if progress.Done {
            truncateStore()
            cancel()
            break
        }

        progressBar.update(progress.Index, progress.Downloaded, progress.Size)
    }
}()

executeCommand(ctx)

ticker := time.NewTicker(time.Second)
defer ticker.Stop()

for {
    select {
    case <-ctx.Done():
        return
    case <-interrupt:
        stopDownload()
        if err := conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")); err != nil {
            log.Println("Error sending close signal to server:", err)
            return
        }

        select {
        case <-ctx.Done():
        case <-time.After(time.Second):
        }

        return
    }
}

客户端是一个 CLI，它将向服务器创建请求。当进度条走完后，就会终止。

这是它的样子

Answer 1

错误

write: broken pipe

通常表示接收方（在本例中为客户端）在服务器写入之前关闭了连接。即使您注意到进度条在初始运行时正常工作，后续请求也可能会遇到在服务器端写入消息和在客户端读取/关闭连接之间存在竞争条件的情况。

您可以采取一些步骤来诊断并可能解决问题：

1. 确保持久连接：检查客户端是否在多个请求之间保持持久的 WebSocket 连接。如果客户端关闭 WebSocket 连接，然后尝试使用相同的关闭连接来进行后续请求，您将遇到此类错误。

2. Graceful Shutdown：确保服务器和客户端都正常关闭连接。您使用

   websocket.CloseMessage

   来指示客户端关闭，但您还应该确保服务器确认这一点并从其端优雅地关闭连接。

3. 错误处理：一旦出现错误，

   progressBar

   函数中当前的错误处理就会退出循环。如果您想确保持久通信，即使面对临时错误，这可能会出现问题。考虑添加更精细的错误处理。

4. 上下文处理：确保与

   ctx

   一起使用的

   websocket.DefaultDialer.DialContext(ctx, ws, nil)

   不会过早取消，从而导致连接终止。您的代码确实建议您在

   cancel

   为

   progress.Done

   时

   true

   上下文，但只需仔细检查应用程序的其余部分以确保不会无意中取消上下文。

5. 日志和调试：在服务器和客户端添加更多日志，特别是在连接启动、消息发送/接收和连接终止方面。这将使您更清楚地了解操作顺序，并可能有助于查明哪里出了问题。

6. 多个连接：确保您不会无意中从客户端创建多个 WebSocket 连接，这可能会导致意外行为。

7. 处理连接关闭：确保您在客户端和服务器端处理 WebSocket 的

   CloseHandler

   ，以优雅地管理连接终止。

8. 检查资源限制：确保您的服务器没有达到任何打开的文件描述符限制或类似的资源限制，这可能会导致连接中断。

9. 检查中间件：仔细检查您与 Go Fiber 一起使用的任何中间件。确保它们不会干扰 WebSocket 连接。

如果您以发出快速连续请求的方式使用 CLI 客户端（例如快速重新运行 CLI 命令），您可能需要添加轻微的延迟或确保之前的 WebSocket 连接已完全关闭开始新的一件事。