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

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

您需要promisify heavyProcess 函数才能成功使用await：

function heavyProcess(message) {
    return new Promise( async (resolve, reject) => {
        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;
        resolve('Done');
    });
}

Answer 2

编辑： 它没有被丢弃的原因是因为你只在做console.log。在它下面添加一个return。

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： 在 cmets 之后：OP 所需的行为是，heavyProcess 应该只发生一次，并且只有在进程完成时才开始一个新的。

现在每个事件都在启动一个新的 HeavyProcess，因为 await 只是等待结果，而不是阻塞。

您必须自己在事件中引入阻塞（或使用 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