按顺序处理websocket消息

Question

我每个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')
}

日志：

1. 启动进程
2. 启动进程
3. 结束进程
4. 结束进程

不幸的是，正如您所看到的，消息继续被处理，而没有等待前一个完成。你有解决这个问题的办法吗？我最后应该使用redis队列来处理消息吗？

谢谢

Answer 1

我不是一个nodeJS的人，但我已经在其他语言中多次考虑过同样的问题。我得出的结论是，消息处理操作的速度有多慢确实很重要，因为如果它们太慢（慢于某个阈值，取决于每秒消息的值），这会导致websocket连接上的瓶颈，而当这个瓶颈形成时，会导致未来消息的极度延迟。

如果await和async具有与python中相同的行为，那么如果您使用它们处理任何操作，您的处理将是异步的，这意味着它确实不会等待前一个操作被处理。

到目前为止,我有两个选择：

1. 保持异步处理消息，但在处理消息的代码中编写一些额外的逻辑，以管理混乱的订单。例如，在继续处理当前消息之前，请确认已经处理了前一条消息。此逻辑可能复杂且缓慢，因为它在单独的线程中运行，并且不会阻塞websocket消息。
2. 逐个同步处理消息，但通过执行一个操作：将消息存储在Redis中，速度极快。这比将它们存储在数据库中要快得多，而且在大多数情况下，速度足够快，不会在WS连接中造成瓶颈。然后使用单独的进程从Redis获取这些消息并对其进行处理。