我每个 WebSocket 收到数十条消息,这些消息的到达时间可能存在几毫秒的差异。我需要通过有时需要一点时间的操作来处理这些数据(例如在数据库中插入)。 为了处理收到的新消息,必须先处理完前一条消息。
我的第一个想法是用node.js Bull(用Redis)准备一个队列,但我担心运行时间太长。这些消息的处理必须保持快速。
我尝试使用 JS 迭代器/生成器(直到现在我从未使用过)并且我测试了如下内容:
const ws = new WebSocket(`${this.baseUrl}${this.path}`)
const duplex = WebSocket.createWebSocketStream(ws, { encoding: 'utf8' })
const messageGenerator = async function* (duplex) {
for await (const message of duplex) {
yield message
}
}
for await (let msg of messageGenerator(socketApi.duplex)) {
console.log('start process')
await this.messageHandler.handleMessage(msg, user)
console.log('end process')
}
日志:
- 启动进程
- 启动进程
- 结束进程
- 结束进程
不幸的是,正如您所看到的,消息会继续处理,而无需等待上一条消息完成。你有解决这个问题的办法吗? 我最终应该使用 Redis 队列来处理消息吗?
谢谢
最佳答案
我不是一个nodeJS人员,但我在其他语言中多次思考过同样的问题。我得出的结论是,消息处理操作的速度有多慢确实很重要,因为如果它们太慢(慢于某个阈值,具体取决于每秒消息的值),这可能会导致 websocket 连接出现瓶颈,并且当该瓶颈出现时向上它可能会导致 future 消息的极度延迟。
如果 await
和 async
具有与 python 中相同的行为,如果您使用它们处理任何操作,您的处理将是异步的,这意味着它确实不会等待以便处理前一个。
到目前为止我有两个选择:
- 继续异步处理消息,但在处理消息的代码中编写一些额外的逻辑,以管理顺序困惑。例如,在继续处理当前消息之前,请确认前一条消息已被处理。此逻辑可能复杂且缓慢,因为它在单独的线程中运行并且不会阻止 websocket 消息。
- 同步处理消息,一条一条,但只需执行一个操作即可实现极快的速度:将它们存储在 Redis 中。这比将它们存储在数据库中要快得多,并且在大多数情况下足够快,不会导致 WS 连接出现瓶颈。然后使用单独的进程从Redis获取这些消息并处理它们。
关于javascript - 如何顺序处理websocket消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/67197842/