javascript - Q - 执行一系列 promise 并在 DAG 中定义它们之间的依赖关系

Question

我想处理一系列数据，其中每个数据的输出都可以用作其他数据的输入。

例如:

    var batch = [
        {"id":"a1","depends":[],"data":{"some":"data a1"}},
        {"id":"b1","depends":["a1"],"data":{"some":"data b1"}},
        {"id":"b2","depends":["a1"],"data":{"some":"data b2"}},
        {"id":"c1","depends":["b1","b2"],"data":{"some":"data c1"}},
        {"id":"x1","depends":[],"data":{"some":"data x1"}},
    ];

这意味着一旦a1完成，它的输出将同时发送到b1和b2；当这些完成时，它们的输出都将被发送到 c1(仅在它们都完成时。x1 可以与 a1、b1、b2 和 c1 并行执行>;和 b1 可以与 b2 并行执行，因为它们之间没有定义 depends。

c1 和 x1 完成后，所有 5 个都完成，应该返回所有 5 个的输出。

我们假设没有定义循环依赖，因此是有向无环图 (DAG)

我想知道如何使用 Q 来实现它，因为:

• 所有数据处理都是异步的，因此我需要使用回调、延迟和 promise ；我更喜欢后者
• Promises 可以作为一种方便的方式来定义图中的边

然而，我还没有能够把它带到概念阶段

var doPromises = {};

var doData = function(data, dependsResultsHash, callback) {
  //Not real processing, simply echoes input after a delay for async simulation purposes
  var out = {
    echo: {
      data: data,
      dependsResultsHash: dependsResultsHash
    }
  };
  setTimeout(function() {
    callback(out);
  }, 1000);
};

var doLine = function(id, depIds, data) {
  var deferred = Q.defer;
  var dependsPromises = [];
  for (var i = 0; i < depIds.length; ++i) {
    var depId = depIds[i];
    dependPromise = doPromises[depId];
    dependsPromises.push(dependPromise);
  }
  Q.all(dependsPromises).then(function(dependsResults) {
    var dependsResultsHash = {};
    for (var i = 0; i < depIds.length; ++i) {
      var depId = depIds[i];
      var depResult = dependsResults[i];
      dependsResultsHash[depId] = depResult;
    }
    doData(data, dependsResultsHash, function(result) {
      deferred.resolve(result);
    });
  });
  return deferred.promise;
}

var doBatch = function(batch) {
  var linePromises = [];
  for (var i = 0; i < batch.length; ++i) {
    var line = batch[i];
    var linePromise = doLine(line.id, line.depends, line.data);
    linePromises.push(linePromise);
    doPromises[line.id] = linePromise;
  }
  Q.all(linePromises).then(function(lineResults) {
    console.log(lineResults);
    deferred.resolve(lineResults);
  });
};

doBatch(batch);

(请注意，此代码未经测试，我不希望它起作用，只是为了说明我的问题所必需的要点。)

我想知道:

• 我这样做对吗？我是否完全忽略了 Q 库的重点。还是延期和 promise ？

• 我主要关心的是 doData 函数:

  -- Is the way that I have selected the promises of the lines depended upon from the global list of promises `doPromises` ok?
  -- Is the way that I have obtained the results of the lines depended upon, and inpterpreted that OK?
  

• 使用doBatch 函数:

  -- I have a local array for `linePromises` and an external hash for `doPromises`, and I feel that these should be combined. How can I do this correctly?
  

• 一般

  -- The code above presently assumes that all `deferred`s will eventually keep their `promise`s. What if they fail or throw an exception; how do I make my code more robust in handling this?
  -- I have used a closure allow acces to `doPromises` in both `doBatch` and `doLine`, and it seems a little odd here, is there a better way to do this?
  

Answer 1

我创建了一个执行此操作的库:

qryq是一个 NodeJs 库，它允许人们表达一系列查询并定义它们之间的依赖关系，无论是并行、顺序还是在有向无环图中。