我每个websocket接收几十条消息,这些消息可能只差几毫秒就能到达。我需要用操作来处理这些数据,这些操作有时会花费一些时间(例如,在DB中的插入)。为了处理接收到的新消息,必须完成对前一个消息的处理。
我的第一个想法是用Node.js Bull(用Redis)准备一个队列,但恐怕太长了,无法运行。这些消息的处理必须保持快速。
我尝试使用JS迭代器/生成器(直到现在我还从未使用过),我测试了如下内容:
const ws = new WebSocket(`${this.baseUrl}${this.path}`)
const duplex = WebSocket.createWebSocketStream(ws, { encoding: 'utf8' })
const messageGenerator = async function* (duplexArg) {
for await (const message of duplexArg) {
yield message
}
}
for await (let msg of messageGenerator(duplex)) {
console.log('start process')
await this.messageHandler.handleMessage(msg)
console.log('end process')
}
日志:
不幸的是,正如您所看到的,消息继续被处理,而没有等待前一个完成。你有解决这个问题的办法吗?我最后应该使用redis队列来处理消息吗?
谢谢
我不是一个nodeJS的人,但我已经在其他语言中多次考虑过同样的问题。我得出的结论是,消息处理操作的速度有多慢确实很重要,因为如果它们太慢(慢于某个阈值,取决于每秒消息的值),这会导致websocket连接上的瓶颈,而当这个瓶颈形成时,会导致未来消息的极度延迟。
如果await
和async
具有与python中相同的行为,那么如果您使用它们处理任何操作,您的处理将是异步的,这意味着它确实不会等待前一个操作被处理。
到目前为止,我有两个选择: