如何实现快速双向消息通信

在使用 Gorilla Websocket 进行双向消息通信时，连续发送多条消息可能导致接收方在一次读取中收到所有消息，而非预期的逐条接收。这是因为接收方假设一次读取只能返回一条消息，而实际上，一条消息可能被拆分多次读取，或者一次读取可能包含多条消息。为了解决这个问题，需要实现一个函数，通过扫描接收到的缓冲区，逐条提取完整的消息。该函数可以不断扫描缓冲区，直至找到一条完整的消息，然后将其返回，同时更新缓冲区的内容。

我有两个服务在单独的 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() 调用中收到一条包含所有数据的消息。

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

我在这里遗漏了一些基本的东西，还是我只需要对发送者/接收者进行额外的调用来一次只处理一条消息？

解决方案

如果消息被缓冲，那么同时收到两条消息是正常的。问题在于接收端假设一次读取返回一条消息。

正如您所见，一次读取可能会返回多条消息。而且，一条消息可能会被拆分为多次读取。后者取决于消息大小。只有您知道消息是什么以及它是如何分隔的。

您必须实现一个返回下一条消息的函数。这是一个建议的实现，假设消息读取的状态存储在结构中。

type MessageParser struct {
    buf []byte
    nBytes int
    conn ... 
}

func NewMessageParser(conn ...) *MessageParser {
    return &MessageParser{
        buf: make([]byte, 256) // best gess of longest message size
        conn: conn
    }
}

func (m *MessageParser) NextMessage() (string, error) {
    var nOpenBrakets, pos int
    var inString bool
    for {
        // scan m.buf to locate next message
        for pos < m.nBytes {
            if m.buf[pos] == '{' && !inString {
                nOpenBrakets++
            } else if m.buf[pos] == '}' && !inString {
                nOpenBrakets--
                if nOpenBrakets == 0 {
                    // we found a full message
                    msg := string(m.buf[:pos+1])
                    m.nBytes = copy(buf, buf[pos+1:m.nBytes)
                    return msg, nil
                }
            } else if m.buf[pos] == '"' {
                if !inString {
                    inString = true
                } else if pos > 0 && m.buf[pos-1] != '\\' {
                    inString = false
                }
            }
            pos++
        }
        // if a message is longer than the buffer capacity, grow the buffer
        if m.nBytes == len(m.buf) {
            temp := make([]byte, len(m.buf)*2)
            copy(temp, m.buf)
            m.buf = temp
        }
        // we didn’t find a full message, read more data
        n, err := conn.Read(m.buf[m.nBytes:]
        m.nBytes += n
        if n == 0 && err != nil {
            return "", err
        }
    }
}

今天关于《如何实现快速双向消息通信》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！