go 例程中的 Websockets：错误上一条消息未读完整

Question

我最近开始探索 Go 并且非常喜欢它。我在尝试检测 WebSocket 连接超时时遇到了问题。我无限期地监听 websocket 连接，当我在 X 秒内没有得到响应时，我尝试重新连接。为了实现这一点，我必须修改我的 for 循环以包含选择。然后我创建了一个类型和一个通道来监听 websocket 响应。然而，这导致我的 websocket 连接出现错误，提示 failed to get reader: previous message not read to Complete。我将从项目中获取的代码替换为独立的。下面是包含两个循环的完整脚本（工作和非工作可用）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            }            buf := new(bytes.Buffer)            buf.ReadFrom(res.Msg)            s := buf.String()            fmt.Println(s)            ticker = time.NewTicker(5 * time.Second)        case <-ticker.C:            fmt.Println("timeout error")            break        }    }}

Answer 1

正如该错误所暗示的那样，必须先完全读取一条消息，然后才能读取下一条消息。使用代码的第一个版本或更改第二个版本以将消息吸收为 [] 字节并将该 [] 字节发送到通道。

假设您正在使用 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

    }

}