goroutine中的Websockets：错误之前的消息未读取完成

Question

我最近开始探索 Go 并且非常喜欢它。我在尝试检测 websocket 连接超时时遇到了问题。我正在无限期地监听 websocket 连接，当我在 X 秒内没有得到响应时，我尝试重新连接。为此，我不得不修改我的 for 循环以包含一个选择。然后我创建了一个类型和一个通道来监听 websocket 响应。然而，这导致我的 websocket 连接出现错误，提示无法获取阅读器：之前的消息未读完。

我将项目中的代码替换为独立的。下面是两个循环的完整脚本（工作和非工作可用）

package main

import (

    "bytes"

    "context"

    "fmt"

    "io"

    "time"

    "nhooyr.io/websocket"

)

func main() {

    ctx := context.Background()

    c, _, err := websocket.Dial(ctx, "wss://stream.binance.com:9443/ws/btcusdt@trade", nil)

    if err != nil {

        fmt.Println(err)

        return

    }

    type wsResponse struct {

        Msg     io.Reader

        Err     error

        MsgType websocket.MessageType

    }

    // THIS LOOP WORKS

    // for {

    //  _, msg, err := c.Reader(ctx)

    //  buf := new(bytes.Buffer)

    //  buf.ReadFrom(msg)

    //  fmt.Println(buf.String())

    //  if err != nil {

    //      fmt.Println(err)

    //      return

    //  }

    // }

    // The following goroutine and loop produces errors

    wsChan := make(chan wsResponse)

    go func() {

        for {

            msgType, msg, err := c.Reader(ctx)

            res := wsResponse{Msg: msg, Err: err, MsgType: msgType}

            //fmt.Printf("%+v\n", res)

            wsChan <- res

        }

    }()

    ticker := time.NewTicker(30 * time.Second)

    for {

        select {

        case res := <-wsChan:

            ticker.Stop()

            if res.Err != nil {

                fmt.Println(res.Err)

                break

            }

        }

    }

}

正在打印日志：

{"e":"trade","E":1577140149102,"s":"BTCUSDT","t":220054947,"p":"7304.40000000","q":"0.07153400","b":933798088 "a":933798124,"T":1577140149099,"m":true,"M":true}

未能获得阅读器：上一条消息未读完

{"e":"trade","E":1577140149107,"s":"BTCUSDT","t":220054948,"p":"7304.95000000","q":"0.28826900","b":933798126 "a":933798125,"T":1577140149104,"m":false,"M":true}

未能获得阅读器：上一条消息未读完

所以它的工作，但它仍然返回错误。阅读器功能源在这里。https://github.com/nhooyr/websocket/blob/master/conn.go#L390。假设我可以在那里提出一个问题。

Answer 1

正如错误所暗示的，必须先完整阅读一条消息，然后才能阅读下一条消息。使用代码的第一个版本或更改第二个版本以将消息发送到 []byte 并将该 []byte 发送到通道。假设您使用的是 nhooyr.io/websocket 包，第二个版本将如下所示：for {&nbsp; &nbsp; // Read returns the entire message as a []byte&nbsp; &nbsp; msgType, msg, err := c.Read(ctx)&nbsp; &nbsp; // bytes.NewReader creates an io.Reader on a []byte&nbsp; &nbsp; res := wsResponse{Msg: bytes.NewReader(msg), Err: err, MsgType: msgType}&nbsp; &nbsp; wsChan <- res&nbsp; &nbsp; if res.Err {&nbsp; &nbsp; &nbsp; &nbsp; // Always exit the loop on error. Otherwise, the goroutine will run forever.&nbsp; &nbsp; &nbsp; &nbsp; return&nbsp; &nbsp; }}