node.js - 如何在事件处理程序中实现异步/等待？

Question

我在事件处理程序中实现 await 时遇到问题。出于某种原因，事件处理程序在开始新进程之前不会等待第一个进程完成。为什么会出现这种奇怪的行为？

const { EventEmitter } = require("events");

let alreadyRunning = false;

const sampleEventHandler = new EventEmitter();
sampleEventHandler.on("data", async (message) => {
  await heavyProcess(message);
});

async function heavyProcess(message) {
  console.log("New message: ", message);
  console.log("already running?: ", alreadyRunning);
  if (alreadyRunning) {
    console.log("Why doesn't it await for first task to complete fully before entering here?");
  }

  const rand = Math.random() * 1000;
  // set var here
  alreadyRunning = true;
  await sleep(rand);
  // unset here
  alreadyRunning = false;
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

// Emit event 5 times
for (let i = 0; i < 5; i++) {
  sampleEventHandler.emit("data", i);
}

Answer 1

编辑:
之所以不被丢弃，是因为你在做console.log只要。在它下面添加一个返回。

async function heavyProcess(message) {
  console.log("New message: ", message);
  console.log("already running?: ", alreadyRunning);
  if (alreadyRunning) {
    console.log("Why doesn't it await for first task to complete fully before entering here?");
     return // Actually stop executing
  }

  const rand = Math.random() * 1000;
  // set var here
  alreadyRunning = true;
  await sleep(rand);
  // unset here
  alreadyRunning = false;
}

编辑2:
遵循评论:OP 所需的行为是，heavyProcess 应该只发生一次，并且只有在进程完成后才开始一个新的。

每个事件现在都开始一个新的重进程，因为等待只是等待结果，它没有阻塞。

您必须在事件中自己引入阻塞(或使用 RXJS 排气映射或类似的东西)。

let isProcessing = false;

sampleEventHandler.on("data", (message) => {
  if (isProcessing) { return; }
  isProcessing = true;
  const resetProcessing = () => isProcessing = false
  heavyProcess(message).then(resetProcessing, resetProcessing); // Reset of both complete and error