Quelle est la meilleure façon de limiter la concurrence lors de l'utilisation de Promise.all() de ES6 ?

Question

J'ai du code qui itère sur une liste qui a été interrogée à partir d'une base de données et qui fait une requête HTTP pour chaque élément de cette liste. Cette liste peut parfois être un nombre relativement important (des milliers), et je voudrais m'assurer que je n'assaille pas un serveur web de milliers de requêtes HTTP simultanées.

Une version abrégée de ce code ressemble actuellement à ceci...

function getCounts() {
  return users.map(user => {
    return new Promise(resolve => {
      remoteServer.getCount(user) // makes an HTTP request
      .then(() => {
        /* snip */
        resolve();
      });
    });
  });
}

Promise.all(getCounts()).then(() => { /* snip */});

Ce code est exécuté sur Node 4.3.2. Pour réitérer, est-ce que Promise.all soit gérée de manière à ce que seul un certain nombre de promesses soient en cours à un moment donné ?

Answer 1

Limite P

J'ai comparé la limitation de la concurrence des promesses avec un script personnalisé, bluebird, es6-promise-pool, et p-limit. Je pense que p-limite a la mise en œuvre la plus simple et la plus dépouillée pour ce besoin. Voir leur documentation .

Exigences

Pour être compatible avec l'async dans l'exemple

• ECMAScript 2017 (version 8)
• Version du nœud > 8.2.1

Mon exemple

Dans cet exemple, nous devons exécuter une fonction pour chaque URL du tableau (comme, peut-être, une demande d'API). Ici, cette fonction est appelée fetchData() . Si nous avions un tableau de milliers d'éléments à traiter, la concurrence serait certainement utile pour économiser les ressources du processeur et de la mémoire.

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

// Example Concurrency of 3 promise at once
const limit = pLimit(3);

let urls = [
    "http://www.exampleone.com/",
    "http://www.exampletwo.com/",
    "http://www.examplethree.com/",
    "http://www.examplefour.com/",
]

// Create an array of our promises using map (fetchData() returns a promise)
let promises = urls.map(url => {

    // wrap the function we are calling in the limit function we defined above
    return limit(() => fetchData(url));
});

(async () => {
    // Only three promises are run at once (as defined above)
    const result = await Promise.all(promises);
    console.log(result);
})();

Le résultat du journal de la console est un tableau des données de réponse de vos promesses résolues.

Answer 2

Si vous savez comment les itérateurs fonctionnent et comment ils sont consommés, vous n'aurez pas besoin d'une bibliothèque supplémentaire, car il peut devenir très facile de construire votre propre concurrence vous-même. Laissez-moi vous montrer :

/* [Symbol.iterator]() is equivalent to .values()
const iterator = [1,2,3][Symbol.iterator]() */
const iterator = [1,2,3].values()

// loop over all items with for..of
for (const x of iterator) {
  console.log('x:', x)

  // notices how this loop continues the same iterator
  // and consumes the rest of the iterator, making the
  // outer loop not logging any more x's
  for (const y of iterator) {
    console.log('y:', y)
  }
}

Nous pouvons utiliser le même itérateur et le partager entre les travailleurs.

Si vous aviez utilisé .entries() au lieu de .values() vous auriez obtenu un tableau 2D avec [[index, value]] ce que je vais démontrer ci-dessous avec une concurrence de 2

const sleep = t => new Promise(rs => setTimeout(rs, t))

async function doWork(iterator) {
  for (let [index, item] of iterator) {
    await sleep(1000)
    console.log(index + ': ' + item)
  }
}

const iterator = Array.from('abcdefghij').entries()
const workers = new Array(2).fill(iterator).map(doWork)
//    ^--- starts two workers sharing the same iterator

Promise.allSettled(workers).then(() => console.log('done'))

L'avantage est que vous pouvez avoir une fonction de générateur au lieu de tout préparer en même temps.

Ce qui est encore plus génial, c'est que vous pouvez faire stream.Readable.from(iterator) dans node (et éventuellement dans les flux whatwg aussi). et avec ReadbleStream transférable, cela rend ce potentiel très utile dans la fonctionnalité si vous travaillez avec des web workers aussi pour les performances

---

Note : le différent de ce comparé à l'exemple async-pool c'est qu'il crée deux travailleurs, de sorte que si l'un d'eux lance une erreur pour une raison quelconque à l'index 5, cela n'empêchera pas l'autre travailleur de faire le reste. Vous passez donc de 2 concurrences à 1 (et cela ne s'arrêtera pas là). Je vous conseille donc d'attraper toutes les erreurs à l'intérieur de l'élément doWork fonction

Answer 3

Notez que Promise.all() ne déclenche pas le début du travail des promesses, la création de la promesse elle-même le fait.

Dans cette optique, une solution serait de vérifier, à chaque fois qu'une promesse est résolue, si une nouvelle promesse doit être lancée ou si l'on a déjà atteint la limite.

Cependant, il n'est vraiment pas nécessaire de réinventer la roue ici. Une bibliothèque que vous pouvez utiliser à cette fin est la suivante es6-promise-pool . A partir de leurs exemples :

// On the Web, leave out this line and use the script tag above instead. 
var PromisePool = require('es6-promise-pool')

var promiseProducer = function () {
  // Your code goes here. 
  // If there is work left to be done, return the next work item as a promise. 
  // Otherwise, return null to indicate that all promises have been created. 
  // Scroll down for an example. 
}

// The number of promises to process simultaneously. 
var concurrency = 3

// Create a pool. 
var pool = new PromisePool(promiseProducer, concurrency)

// Start the pool. 
var poolPromise = pool.start()

// Wait for the pool to settle. 
poolPromise.then(function () {
  console.log('All promises fulfilled')
}, function (error) {
  console.log('Some promise rejected: ' + error.message)
})

Answer 4

Au lieu d'utiliser des promesses pour limiter les requêtes http, utilisez la fonction intégrée de node http.Agent.maxSockets . Cela supprime la nécessité d'utiliser une bibliothèque ou d'écrire votre propre code de mise en commun, et présente l'avantage supplémentaire d'un meilleur contrôle sur ce que vous limitez.

agent.maxSockets

Par défaut, il est réglé sur Infinity. Détermine le nombre de sockets concurrents que l'agent peut avoir ouverts par origine. L'origine est soit une combinaison 'host:port' ou 'host:port:localAddress'.

Par exemple :

var http = require('http');
var agent = new http.Agent({maxSockets: 5}); // 5 concurrent connections per origin
var request = http.request({..., agent: agent}, ...);

Si vous effectuez plusieurs requêtes auprès de la même origine, il peut être avantageux de définir le paramètre keepAlive à true (voir la documentation ci-dessus pour plus d'informations).

Answer 5

De l'oiseau bleu Promise.map peut prendre une option de concurrence pour contrôler combien de promesses doivent être exécutées en parallèle. Parfois, il est plus facile que .all car vous n'avez pas besoin de créer le tableau de promesses.

const Promise = require('bluebird')

function getCounts() {
  return Promise.map(users, user => {
    return new Promise(resolve => {
      remoteServer.getCount(user) // makes an HTTP request
      .then(() => {
        /* snip */
        resolve();
       });
    });
  }, {concurrency: 10}); // <---- at most 10 http requests at a time
}