javascript - 如何确保在流处理完成后执行异步代码？

Question

我有一个流，我通过监听 data、error 和 end 事件来处理，我调用一个函数来处理第一个流中的每个 data 事件。自然地，处理数据的函数调用其他回调，使其异步。那么如何在处理完流中的数据后开始执行更多的代码呢？在流中监听 end 事件并不意味着异步 data 处理函数已经完成。

如何确保在执行下一条语句时流数据处理函数已经完成？

这是一个例子:

function updateAccountStream (accountStream, callThisOnlyAfterAllAccountsAreMigrated) {
  var self = this;
  var promises = [];
  accountStream
    .on('data', function (account) {
      migrateAccount.bind(self)(account, finishMigration);
    })
    .on('error', function (err) {
      return console.log(err);
    })
    .on('end', function () {
      console.log("Finished updating account stream (but finishMigration is still running!!!)");
      callThisOnlyAfterAllAccountsAreMigrated() // finishMigration is still running!
    });
}

var migrateAccount = function (oldAccount, callback) {
  executeSomeAction(oldAccount, function(err, newAccount) {
    if (err) return console.log("error received:", err);
    return callback(newAccount);
  });
}

var finishMigration = function (newAccount) {
  // some code that is executed asynchronously...
}

如何确保在处理流后调用 callThisOnlyAfterAllAccountsAreMigrated？

这可以用 promise 来完成吗？可以通过流来完成吗？我正在使用 Nodejs，因此引用其他 npm 模块可能会有所帮助。

Answer 1

如您所说，监听流上的 end 事件本身是无用的。流不知道也不关心您对 data 处理程序中的数据做了什么，因此您需要编写一些代码来跟踪您自己的 migrateAccount 状态。

如果是我，我会重写整个部分。如果您在流中使用带有 .read() 的 readable 事件，您可以一次读取任意数量的项目。如果是这样，没问题。如果是30，那很好。你这样做的原因是你不会超限任何正在处理来自流的数据的工作。就目前而言，如果 accountStream 很快，您的应用程序无疑会在某个时候崩溃。

当您从流中读取一个项目并开始工作时，接受您返回的 promise (使用 Bluebird 或类似工具)并将其放入一个数组中。解决 promise 后，将其从数组中删除。当流结束时，将 .done() 处理程序附加到 .all()(基本上是从数组中的每个 promise 中做出一个大 promise )。

您还可以对正在进行的作业使用一个简单的计数器。