如何快速连续发送两条消息，一次全部收到

Question

我有两个服务在单独的 Docker 容器中运行，它们使用 Gorilla Websocket 在彼此之间发送消息。我可以一次发送一个很好的消息，但是当我快速连续发送两个消息时，它们在一次读取期间到达接收器，导致我的解组失败。

在发送方我有一个循环发送两条消息：

for _, result := range results {

    greetingMsg := Message{

        TopicIdentifier: *bot.TopicIdentifier,

        UserIdentifier:  botIdentifier,

        Message:         result,

    }

    msgBytes, err := json.Marshal(greetingMsg)

    if err != nil {

        log.Println("Sender failed marshalling greeting message with error " + err.Error())

    }

    log.Printf("Sender writing %d bytes of message\n%s\n", len(msgBytes), string(msgBytes))

    err = conn.WriteMessage(websocket.TextMessage, msgBytes)

    if err != nil {

        log.Printf("Sender failed to send message\n%s\nwith error %s ", string(msgBytes), err.Error())

    }

}

正如预期的那样，我在 conn.WriteMessage() 调用之前得到了两个日志：

2019/12/12 06:23:29 agent.go:119: Sender writing 142 bytes of message

{"topicIdentifier":"7f7d12ea-cee8-4f05-943c-2e802638f075","userIdentifier":"753bcb8a-d378-422e-8a09-a2528565125d","message":"I am doing good"}

2019/12/12 06:23:29 agent.go:119: Sender writing 139 bytes of message

{"topicIdentifier":"7f7d12ea-cee8-4f05-943c-2e802638f075","userIdentifier":"753bcb8a-d378-422e-8a09-a2528565125d","message":"How are you?"}

在接收端，我正在收听如下：

_, msg, err := conn.ReadMessage()

fmt.Printf("Receiver received %d bytes of message %s\n", len(msg), string(msg))

并且该日志消息会产生：

2019/12/12 06:23:29 Receiver received 282 bytes of message  {"topicIdentifier":"83892f58b4b0-4303-8973-4896eed67ce0","userIdentifier":"119ba709-77a3-4b34-92f0-2187ecab7fc5","message":"I am doing good"}

{"topicIdentifier":"83892f58-b4b0-4303-8973-4896eed67ce0","userIdentifier":"119ba709-77a3-4b34-92f0-2187ecab7fc5","message":"How are you?"}

因此，对于发送方的两个 conn.WriteMessage() 调用，我在接收方的 conn.ReadMessage() 调用中收到一条消息，其中包含所有数据。

我认为这里存在某种竞争条件，因为有时接收者确实会按预期收到两条单独的消息，但这种情况很少发生。

我在这里是否缺少一些基本的东西，或者我只需要对发送者/接收者进行额外的调用以一次只处理一条消息？

Answer 1

如果消息被缓冲，则两个消息同时被接收是正常的。问题出在接收端，它假设一次读取返回一条消息。如您所见，一次阅读可能会返回多条消息。而且，一条消息可能会被拆分为多次读取。后者取决于消息大小。只有您知道消息是什么，以及它是如何定界的。您必须实现一个返回下一条消息的函数。这是一个建议的实现，假设消息读取的状态存储在结构中。type MessageParser struct {&nbsp; &nbsp; buf []byte&nbsp; &nbsp; nBytes int&nbsp; &nbsp; conn ...&nbsp;}func NewMessageParser(conn ...) *MessageParser {&nbsp; &nbsp; return &MessageParser{&nbsp; &nbsp; &nbsp; &nbsp; buf: make([]byte, 256) // best gess of longest message size&nbsp; &nbsp; &nbsp; &nbsp; conn: conn&nbsp; &nbsp; }}func (m *MessageParser) NextMessage() (string, error) {&nbsp; &nbsp; var nOpenBrakets, pos int&nbsp; &nbsp; var inString bool&nbsp; &nbsp; for {&nbsp; &nbsp; &nbsp; &nbsp; // scan m.buf to locate next message&nbsp; &nbsp; &nbsp; &nbsp; for pos < m.nBytes {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if m.buf[pos] == '{' && !inString {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; nOpenBrakets++&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else if m.buf[pos] == '}' && !inString {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; nOpenBrakets--&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if nOpenBrakets == 0 {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // we found a full message&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; msg := string(m.buf[:pos+1])&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; m.nBytes = copy(buf, buf[pos+1:m.nBytes)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return msg, nil&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else if m.buf[pos] == '"' {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; if !inString {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; inString = true&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; } else if pos > 0 && m.buf[pos-1] != '\\' {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; inString = false&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; pos++&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; // if a message is longer than the buffer capacity, grow the buffer&nbsp; &nbsp; &nbsp; &nbsp; if m.nBytes == len(m.buf) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; temp := make([]byte, len(m.buf)*2)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; copy(temp, m.buf)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; m.buf = temp&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; // we didn’t find a full message, read more data&nbsp; &nbsp; &nbsp; &nbsp; n, err := conn.Read(m.buf[m.nBytes:]&nbsp; &nbsp; &nbsp; &nbsp; m.nBytes += n&nbsp; &nbsp; &nbsp; &nbsp; if n == 0 && err != nil {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return "", err&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }}

Answer 2

如果您查看写入功能的 gorilla WebSockets 代码NextWriter returns a writer for the next message to send. The writer's Close// method flushes the complete message to the network.读者也有相同的实现。它似乎是正确的。也许正如@chmike 所建议的那样，消息可能已经被缓冲了。至于实现，您始终可以在消息末尾添加分隔符，并在阅读时解析消息直到到达分隔符（以防消息溢出）func writeString(conn *websocket.Conn, data []byte) {conn.WriteMessage(1, append(data, "\r\n"...))}我试图重现相同的内容，但它对我不起作用。在低级别，连接api通常使用c文件编译。您可以尝试使用“-tags netgo”构建您的应用程序，以纯粹使用 go 构建它。