Node.js `Stream` 和 block 顺序

Question

Node.js stream 的顺序是否保留到并包括 'data' 事件？我相信他们是，并且this question似乎也至少表明 pipe 方法是顺序的。

我希望保留 block 顺序，以便我可以在流结束之前调用一个对每个 block 进行操作的函数，即，在读取流的一部分时，我希望缓冲的数据 block 由'data' 事件中的函数。

但是，当我运行以下代码时，我使用 Math.random 和 setTimeout 来专门测试事件是否按顺序发生:

fs.createReadStream(filePath, { encoding: 'ascii' })
  .pipe(streamToEntry)
  .on('data', (chunk) => {
    setTimeout(() => { 
      console.log(chunk); 
    }, Math.random() * 1000)
  });

数据 block 可以以无序的方式记录。

这是因为 setTimeout() 还是因为 'data' 事件不一定按顺序调用？即，有序处理应该只发生在 pipe 方法中，还是我可以在最后按顺序处理数据？

Answer 1

保证您的data事件按顺序发出。请参阅this answer有关一些其他详细信息和 Node 源代码中的一些片段，这表明您确实会按顺序获取 data 事件。

当您在 data 回调中添加异步代码时，就会出现此问题(setTimeout 是异步代码的示例)。在这种情况下，无法保证您的data回调按照调用顺序完成处理。

您需要做的是确保在 data 回调返回时，您已完全处理数据。换句话说，您的回调代码需要是同步代码。

fs.createReadStream(filePath, { encoding: 'ascii' })
  .pipe(streamToEntry)
  .on('data', (chunk) => {
      // only synchronous code here
      console.log(chunk); 
  });

为了使问题代码正常工作，可以使用async/await:

fs.createReadStream(filePath, { encoding: 'ascii' })
  .pipe(streamToEntry)
  .on('data', async (chunk) => {
      // only synchronous code here
      await setTimeout(() => console.log(chunk), Math.random() * 1000); 
  });