如何按传入顺序处理 socket.io 事件答案

【问题标题】：How to process socket.io events in their incoming order如何按传入顺序处理 socket.io 事件
【发布时间】：2023-03-14 18:11:01
【问题描述】：

我有以下设置：

async MyFunction(param) {
    //... Do some computation
    await WriteToDB()
}

io.on('connection', (socket) => {
    socket.on('AnEvent', (param) => MyFunction(param))
})

当一个事件进来时，它会调用一个异步函数来进行一些计算，最后通过另一个异步调用将结果写入数据库。

如果MyFunction最后没有异步调用写数据库，例如

MyFunction(param) {
    //... Do some computation
}

那么显然所有事件都将按照它们的传入顺序进行处理。下一个事件的处理只有在前一个事件的处理完成后才会开始。但是，由于是异步调用数据库，不知道那些传入的事件是否还会被完全按顺序处理。恐怕下一个事件的处理在前一个await WriteToDB() 完成之前开始。如何更改代码以按顺序完全处理它们？

【问题讨论】：

标签： javascript node.js socket.io

【解决方案1】：

您是正确的，不能保证传入事件将按顺序处理。

要实现您的要求，您需要一个“消息队列”，它会定期检查新消息并一一处理。

const messageQueue = [];

// SocketIO adding Message to MessageQueue
const eventHandler = (message) => {
  messageQueue.push(message);
}

const messageHandler = () => {
  if (messageQueue.length === 0) {
    return;
  }

  const message = messageQueue.shift();

  // Handle Message

  // If successful, ask for next message
  return messageHandler();
}

当然，我的示例非常幼稚，但我希望它能让您大致了解您所要求的内容是如何完成的。

如果您发现自己需要更强大的消息队列，请查看 RabbitMQ、BullMQ、Kafka

【讨论】：

如何调用messageHandler？看起来需要多次调用它。它会在消费完所有消息后返回，因此需要在新消息到来时调用它。
你的Socket IO消息处理程序（从socket io接收消息的处理程序）只需要将消息推送到消息队列中，然后，你的应用程序中的另一个组件（一般由消息队列提供）定期检查队列中的新消息（示例中为 messageQueue 数组），如果有新消息，它将处理它。
我强烈建议您研究 BullMQ、RabbitMQ、Kafka 的工作原理，您的答案以及所有未来的问题很可能在他们的文档中得到解答。