为什么事件侦听器“暂停”执行异步函数以开始新的执行？

Question

如果第二次发出相同的事件，为什么 Node.js 事件侦听器会“暂停”异步函数的执行以开始第二次执行？第二个问题：如何完成第一次执行，然后开始第二次执行？

即，如果在 Node.js 中启动此代码：

import { EventEmitter } from "events";

let event = new EventEmitter();

event.on("myEvent", async function () {
  console.log("Start");
  await new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(console.log("Do job"));
    }, 1000);
  });
  console.log("Finish");
});

event.emit("myEvent"); // first emit
event.emit("myEvent"); // second emit

然后我得到这样的结果：

Start
Start
Do job
Finish
Do job
Finish

但是我想看到这个：

Start
Do job
Finish
Start
Do job
Finish

更新 下面我放了包含所描述问题的真实代码

const web3 = new Web3(
  new Web3.providers.WebsocketProvider(
    "wss://eth-mainnet.g.alchemy.com/v2/<API-KEY>"
  )
);
let walletAddress = "0x123";

let options = {
  topics: [web3.utils.sha3("Transfer(address,address,uint256)")],
};

let subscription = web3.eth.subscribe("logs", options);

subscription.on("data", async (event) => {
  if (event.topics.length == 3) {
    let transaction = decodeTransaction(event); //just using web3.eth.abi.decodeLog(...)
    if (
      transaction.from === walletAddress ||
      transaction.to === walletAddress
    ) {
      const contract = new web3.eth.Contract(abi, event.address);
      let coinSymbol = await contract.methods.symbol().call(); //<-- The issue starts here
      await redisClient.hSet(
        walletAddress,
        coinSymbol,
        transaction.value
      );
    }
  }
});

Answer 1

这里的关键是async 函数不会阻塞解释器，EventEmitter 事件不会等待async 事件处理程序解决它们的承诺。

所以，这就是发生的事情：

1. 第一个event.emit() 被调用。这会同步触发 myEvent 处理函数被调用。
2. 该函数执行。在输出start 后，它会命中await。这会导致它暂停函数的进一步执行，并立即将承诺返回给调用者。这会导致第一个event.emit(...) 被完成，因为eventEmitter 对象不是promise-aware - 它根本不注意您的事件处理函数返回的promise。
3. 第二个event.emit() 被调用。这会同步触发 myEvent 处理函数被调用。
4. 该函数执行。输出start 后，它命中await。这会导致它暂停函数的进一步执行，并立即将承诺返回给调用者。这会导致第二个event.emit(...) 完成。

   因此，这就是您的输出以以下开头的原因：

   Start
   Start
   

   然后，稍后（在setTimeout() 触发之后），console.log("Do job") 输出承诺得到解决，这导致await 得到满足，并且函数在await 之后恢复执行。然后输出Finish。

   所以，此时，第一个计时器已经触发，你有：

   Start
   Start
   Do job
   Finish
   

   然后第二个setTimeout() 触发并执行相同的操作，然后您就拥有了完整的输出：

   Start
   Start
   Do job
   Finish
   Do job
   Finish
   

   这里的关键是 EventEmitter 类对于它的事件处理程序来说不是承诺感知的。它不注意您的 async 函数返回的承诺，因此在允许其余代码继续执行之前不会“等待”它解决。

   ---

   如果您在subscription.on('data', ...) 代码中尝试执行的操作是序列化每个事件的处理，以便在开始处理下一个事件之前完成处理一个事件，那么您可以将事件排队并仅在前一个已经完成。如果在您仍在处理前一个事件时有一个新事件到达，它只会被放入队列中并一直留在那里，直到前一个事件完成处理。

   以下是该代码的外观：

   const eventQueue = [];
   let eventInProgress = false;
   
   async function processEvent(event) {
       try {
           eventInProgress = true;
           if (event.topics.length == 3) {
               let transaction = decodeTransaction(event); //just using web3.eth.abi.decodeLog(...)
               if (
                   transaction.from === walletAddress ||
                   transaction.to === walletAddress
               ) {
                   const contract = new web3.eth.Contract(abi, event.address);
                   let coinSymbol = await contract.methods.symbol().call(); //<-- The issue starts here
                   await redisClient.hSet(
                       walletAddress,
                       coinSymbol,
                       transaction.value
                   );
               }
           }
       } catch (e) {
           // have to decide what to do with rejections from either of the await statements
           console.log(e);
       } finally {
           eventInProgress = false;
           // if there are more events to process, then process the oldest one
           if (eventQueue.length) {
               processEvent(eventQueue.shift());
           }
       }
   }
   
   subscription.on("data", (event) => {
       // serialize the processing of events
       if (eventInProgress) {
           eventQueue.push(event);
       } else {
           processEvent(event);
       }
   });