我有两个服务在单独的 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
}
}
}
关于go - 如何快速连续发送两条消息,一次全部收到,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59299473/