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

Question

我最近开始探索 Go 并且非常喜欢它。我在尝试检测 websocket 连接超时时遇到了问题。我正在无限期地监听 websocket 连接，当我在 X 秒内没有得到响应时，我尝试重新连接。为此，我不得不修改我的 for 循环以包含一个选择。然后我创建了一个类型和一个通道来监听 websocket 响应。然而，这导致我的 websocket 连接出现错误，提示无法获取阅读器：之前的消息未读完。我将项目中的代码替换为独立的。下面是两个循环的完整脚本（工作和非工作可用）package mainimport (    "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 {

    // Read returns the entire message as a []byte

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

    // bytes.NewReader creates an io.Reader on a []byte

    res := wsResponse{Msg: bytes.NewReader(msg), Err: err, MsgType: msgType}

    wsChan <- res

    if res.Err {

        // Always exit the loop on error. Otherwise, the goroutine will run forever.

        return

    }

}