如何限制 Q Promise 并发？答案

【问题标题】：How can I limit Q promise concurrency?如何限制 Q Promise 并发？
【发布时间】：2013-12-31 00:48:30
【问题描述】：

如何编写限制 Q Promise 并发的方法？

例如，我有一个方法spawnProcess。它返回一个 Q 承诺。
我希望一次生成的进程不超过 5 个，但对调用代码透明。

我需要实现的是一个带有签名的函数

function limitConcurrency(promiseFactory, limit)

我可以这样称呼

spawnProcess = limitConcurrency(spawnProcess, 5);

// use spawnProcess as usual

我已经开始研究我的版本，但我想知道是否有人有我可以检查的简洁实现。

【问题讨论】：

您是为浏览器编写代码，还是为 Node 编写代码？如果是前者，就没有并发……
@Matt：我正在为节点写作。我不是指线程中的并发，我指的是“同时运行的承诺”中的并发。
你尝试了什么？使用等待队列和缓冲请求应该不会太难。
@schlingel：我并不是说这很难。准备好后，我将发布完整的代码。我只是想弄清楚如何正确链接承诺，以便下一个开始只要前一个完成。
如果请求可以立即执行，您可以返回由进程解决的延迟，如果您必须将其排队，则必须使用两个单独的延迟。一个调用一个减少计数器的函数并另外解决第二个延迟的函数。第二个 deferred 应该由 spawnProcess 函数返回。

标签： javascript node.js concurrency promise q

【解决方案1】：

我有一个库可以为你做这件事https://github.com/ForbesLindesay/throat

您可以通过 browserify 使用它或从 brcdn (https://www.brcdn.org/?module=throat&version=latest) 下载独立构建并将其添加为脚本标签。

然后（假设 Promise 构造函数在您的环境中被填充或实现）您可以这样做：

//remove this line if using standalone build
var throat = require('throat');

function limitConcurrency(promiseFactory, limit) {
  var fn = throat(promiseFactory, limit);
  return function () {
    return Q(fn.apply(this, arguments));
  }
}

您可以直接调用throat(promiseFactory, limit)，但这将返回promise 承诺而不是Q 承诺。

我也很喜欢将它与 array.map 一起使用。

// only allow 3 parallel downloads
var downloadedItems = Q.all(items.map(throat(download, 3)));

【讨论】：

太好了，谢谢！你的库看起来比我的更像惯用的 JS。还有，测试。
很高兴你发现它有用:)
我更喜欢使用常用库而不是自定义代码的解决方案。 Q 是否有提供此功能的替代承诺库？
throat 是常用的库。如果只是使用node内置的promise实现，不需要任何自定义代码，只需throat(myFunction, parallelism)

【解决方案2】：

这似乎对我有用。

我不确定是否可以简化它。 scheduleNextJob 中的递归是必要的，因此 running < limit 和 limit++ 始终在同一个刻度中执行。

Also available as a gist.

'use strict';

var Q = require('q');

/**
 * Constructs a function that proxies to promiseFactory
 * limiting the count of promises that can run simultaneously.
 * @param promiseFactory function that returns promises.
 * @param limit how many promises are allowed to be running at the same time.
 * @returns function that returns a promise that eventually proxies to promiseFactory.
 */
function limitConcurrency(promiseFactory, limit) {
  var running = 0,
      semaphore;

  function scheduleNextJob() {
    if (running < limit) {
      running++;
      return Q();
    }

    if (!semaphore) {
      semaphore = Q.defer();
    }

    return semaphore.promise
      .finally(scheduleNextJob);
  }

  function processScheduledJobs() {
    running--;

    if (semaphore && running < limit) {
      semaphore.resolve();
      semaphore = null;
    }
  }

  return function () {
    var args = arguments;

    function runJob() {
      return promiseFactory.apply(this, args);
    }

    return scheduleNextJob()
      .then(runJob)
      .finally(processScheduledJobs);
  };
}

module.exports = {
  limitConcurrency: limitConcurrency
}

【讨论】：

【解决方案3】：

Deferred 承诺实现具有 gate 函数，它的工作原理就是这样：

spawnProcess = deferred.gate(spawnProcess, 5);

【讨论】：

【解决方案4】：

我为此编写了一个小库：https://github.com/suprememoocow/qlimit

它非常易于使用，专为与 Q Promise 一起使用而设计：

var qlimit = require('qlimit');
var limit = qlimit(2); // 2 being the maximum concurrency

// Using the same example as above
return Q.all(items.map(limit(function(item, index, collection) { 
  return performOperationOnItem(item);
}));

它也可以用来限制特定资源的并发，像这样：

var qlimit = require('qlimit');
var limit = qlimit(2); // 2 being the maximum concurrency

var fetchSomethingFromEasilyOverwhelmedBackendServer = limit(function(id) {
  // Emulating the backend service
  return Q.delay(1000)
    .thenResolve({ hello: 'world' }); 
});

【讨论】：