我每个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* (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连接出现瓶颈，当这个瓶颈出现时，可能会导致未来消息的极度延迟。
如果await和async的行为与python中的行为相同，那么如果使用它们处理任何操作，则处理将是异步的，这意味着它确实不会等待前一个操作被处理。
到目前为止，我有两个选择：
- 保持异步处理消息，但在处理消息的代码中编写一些额外的逻辑，以管理订单混乱。例如，在继续处理当前消息之前，请确认上一条消息已被处理。这种逻辑可能复杂而缓慢，因为它在一个单独的线程中运行，并且不会阻塞websocket消息
- 同步处理消息，一个接一个，但只需执行一个操作即可快速处理：将消息存储在Redis中。这比将它们存储在数据库中要快得多，而且在大多数情况下速度足够快，不会在WS连接中造成瓶颈。然后使用单独的流程从Redis获取这些消息并进行处理