限制并发并等到所有承诺完成，即使有些拒绝

Posted 2023-03-07

【中文标题】限制并发并等到所有承诺完成，即使有些拒绝

【英文标题】：limit concurrency and wait until all promises complete even if some reject

【发布时间】：2021-03-03 23:55:26

【问题描述】：

我的问题基本上是综合

What is the best way to limit concurrency? Wait until all promises complete even if some rejected

我知道Promise.allSettled，但我没能找到同时限制并发的好方法。

到目前为止我所拥有的：

想法 1 使用p-limit：

const pLimit = require('p-limit');
const limit = pLimit(10);

let promises = files.map(pair => 
    var formData = 
        'file1': fs.createReadStream(pair[0]),
        'file2': fs.createReadStream(pair[1])
    ;
        
    return limit(() => uploadForm(formData));
);
    
(async () => 
    const result = await Promise.allSettled(promises).then(body => 
        body.forEach(value => 
            if(value.status == "rejected")
                file.write(value.reason + '\n---\n');
        );
    );
)();

我对这个解决方案的问题是，我必须首先创建所有 Promise，然后为每个 Promise 打开两个文件流，我将达到打开文件的限制。

想法 2 使用p-queue： 我尝试使用生成器函数在 queue.on 'next' 事件中创建和添加新的 Promise，但我无法让它正常工作，这可能不是这项工作的正确工具。

想法 3 使用 PromisePool： 这在一开始看起来很有希望。其中一些支持生成器函数来为池创建 Promise，但我找不到一个明确声明其行为类似于 Promise.allSettled 的函数。

我实现了es6-promise-pool 却发现它会在第一次promise 被拒绝后停止。

【参考方案1】：

自己实现它很简单——创建一个函数数组，在调用时返回 Promise。然后实现一个限制器函数，从该数组中获取函数并调用它们，一旦完成，再次递归调用限制器，直到数组为空：

const request = (file) => new Promise((res, rej) => 
  console.log('requesting', file);
  setTimeout(() => 
    if (Math.random() < 0.5) 
      console.log('resolving', file);
      res(file);
     else 
      console.log('rejecting', file);
      rej(file);
    
  , 1000 + Math.random() * 1000);
);
const files = [1, 2, 3, 4, 5, 6];

const makeRequests = files.map(file => () => request(file));
const results = [];
let started = 0;
const recurse = () => 
  const i = started++;
  const makeRequest = makeRequests.shift();
  return !makeRequest ? null : Promise.allSettled([makeRequest()])
    .then(result => 
      results[i] = result[0];
      return recurse();
    )
;
const limit = 2;
Promise.all(Array.from( length: limit , recurse))
  .then(() => 
    console.log(results);
  );

如果结果的顺序无关紧要，可以通过删除started 和i 变量来简化。

【讨论】：

我喜欢这个解决方案。我从中学到了一些东西。非常感谢。

【参考方案2】：

接受的答案或多或少类似于p-limit。 你遇到了p-limit 的问题，因为流是在限制回调之外声明的。

这会解决你的问题：

let promises = files.map(pair =>   
    return limit(() => uploadForm(
        'file1': fs.createReadStream(pair[0]),
        'file2': fs.createReadStream(pair[1])
    ));
);

【讨论】：

我想这样做，但是如果不将整个应用程序更改为 ES 模块并将所有需要更改为导入，我无法弄清楚如何使用 p-limit。 :(

@JasonC 尝试使用旧版本的 p-limit。通常 esm 的添加会附带一个新专业，请尝试以前的专业。