我有兩個服務在單獨的 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() 調用中收到一條消息，其中包含所有數據。我認為這里存在某種競爭條件，因為有時接收者確實會按預期收到兩條單獨的消息，但這種情況很少發生。我在這里是否缺少一些基本的東西，或者我只需要對發送者/接收者進行額外的調用以一次只處理一條消息？
2 回答
RISEBY
TA貢獻1856條經驗 獲得超5個贊
如果消息被緩沖，則兩個消息同時被接收是正常的。問題出在接收端，它假設一次讀取返回一條消息。
如您所見，一次閱讀可能會返回多條消息。而且，一條消息可能會被拆分為多次讀取。后者取決于消息大小。只有您知道消息是什么，以及它是如何定界的。
您必須實現一個返回下一條消息的函數。這是一個建議的實現，假設消息讀取的狀態存儲在結構中。
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
}
}
}
慕姐8265434
TA貢獻1813條經驗 獲得超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 構建它。
- 2 回答
- 0 關注
- 400 瀏覽
添加回答
舉報
0/150
提交
取消