javascript - 使用异步处理高原流 block

Question

我正在使用 highland.js 来处理文件，使用流读取两个分隔符之间的内容。我还使用 async.js 按顺序运行一系列 http 请求。

理想情况下，我希望将 highland 的输出 x 作为第一个函数传递给 async 系列(链)，以便为提取的每个 block 执行 HTTP 请求来自流。

这可能吗？如果可以，如何实现？

var async = require('async');
var _ = require('highland');


_(fs.createReadStream(files[0], { encoding: 'utf8' }))
        .splitBy('-----BEGIN-----\n')
        .splitBy('\n-----END-----\n')
        .filter(chunk => chunk !== '')
        .each(function (x) {
        }).done(function () {

    async.series([
        function(callback) {
            setTimeout(function() {
                console.log('Task 1');
                callback(null, 1);
            }, 300);
        },
        function(callback) {
            setTimeout(function() {
                console.log('Task 2');
                callback(null, 2);
            }, 200);
        },
    ], function(error, results) {
        console.log(results);
    });

});;

Answer 1

您可以摆脱对 each 和 done 的调用。过滤后，可以用.toArray(callback)跟进。向回调传递一个包含高地结果的数组。你可以像这样重构

var Q = require('q');
var _ = require('highland');


_(fs.createReadStream(files[0], { encoding: 'utf8' }))
        .splitBy('-----BEGIN-----\n')
        .splitBy('\n-----END-----\n')
        .filter(chunk => chunk !== '')
        .each(asyncTasks);

function asyncTasks(x) { // here, x will be each of the results from highland
    async.series([
      // do something with x results
        function(callback) {
          console.log('Task 1');
          callback(null, 1);
        },
        // do something else with x results
        function(callback) {
          console.log('Task 2');
          callback(null, 2);
        },
    ], function(error, results) {
        console.log(results);
    });
}

here是 toArray 文档的链接。 toArray 消耗流，就像 done 一样。如果您有任何疑问，请告诉我。

尽管说实话，我认为使用 Promise 会更好。虽然部分只是个人喜好，部分是因为它使代码更具可读性。来自 what I've read ，异步比 Promise 性能更高，但 Promise 的好处是您可以将结果从一个函数传递到下一个函数。因此，在您的示例中，您可以在第一部分中对 x 执行一些操作，然后将修改后的结果传递给下一个函数，然后再传递给下一个函数，依此类推。当您使用 async.series 时，您可以通过调用 callback(null, result) 来完成每个函数，并且只有在系列的最后完成时(当您从所有对 callback 的调用中获取所有结果时)才能获得结果。现在，您始终可以将结果保存到 async.series 之外的某个变量中，但这会使您的代码更加困惑。如果你想用 Promise 重写它，它会如下所示。我在这里使用 q，但这只是您可以使用的众多 Promise 库之一。

    var async = require('async');
    var _ = require('highland');


    _(fs.createReadStream(files[0], { encoding: 'utf8' }))
            .splitBy('-----BEGIN-----\n')
            .splitBy('\n-----END-----\n')
            .filter(chunk => chunk !== '')
            .each(asyncTasks);

    function asyncTasks(x) { // here, x will be an array of the results from highland
      return asyncTask1(x)
              .then(asyncTask2)
              .then(asyncTask3)
    }

    function asyncTask1(x) {
      var deferred = Q.defer();

      // do some stuff

      if (// some error condition) {
        deferred.reject();
      } else {
        deferred.resolve(x); // or pass along some modified version of x
      }

      return deferred.promise;
    }

    function asyncTask2(x) {
      // same structure as above
    }

    function asyncTask3(x) {
      // same structure as above
    }

现在一些异步 API 除了接受回调之外，还开始返回 Promise，或者有时代替。所以能够感到舒服是一件好事。 promise 非常有用。您可以阅读有关它们的更多信息 here和 here .