用异步技术处理高地流块

Question

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

理想情况下，我希望将来自highland的输出x作为第一个函数传递给async系列(链)，以便对从流中提取的每个块执行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);
    });
}

这里是指向toArray文档的链接。toArray消耗流，就像done也使用流一样。如果你有任何问题，请告诉我。

不过老实说，我认为你最好还是用承诺代替。虽然它的一部分只是个人偏好，部分原因在于它使代码更具可读性。在我所读到的中，异步比承诺更具有性能，但是承诺的好处在于您可以将结果从一个函数传递到下一个函数。因此，在您的示例中，您可以在第一部分中对x执行一些操作，然后将修改后的结果传递给下一个函数和下一个函数，以此类推。当您使用async.series时，您通过调用callback(null, result)来完成每个函数，直到在本系列的最后得到所有对callback的调用的结果之后，才会得到结果。现在，您可以将结果保存到async.series之外的某个变量，但这会使您的代码变得更加混乱。如果你想用承诺重写它，它将如下所示。我在这里使用q，但它只是您可以使用的许多承诺库之一。

    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已经开始返回承诺，除了接受回调，或者有时代替。所以这将是一件很好的事情，让人感到舒服。承诺是非常有用的。您可以阅读更多关于它们的这里和这里。