Was ist der beste Weg, um die Parallelität zu begrenzen, wenn Sie Promise.all () von ES6 verwenden?

Ich habe einen Code, der über eine Liste iteriert, die aus einer Datenbank abgefragt wurde, und eine HTTP-Anforderung für jedes Element in dieser Liste stellt. Diese Liste kann manchmal eine relativ große Zahl sein (zu Tausenden), und ich möchte sicherstellen, dass ich keinen Webserver mit Tausenden von gleichzeitigen HTTP-Anforderungen erreiche.

Eine abgekürzte Version dieses Codes sieht derzeit ungefähr so ​​aus ...

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 */});

Dieser Code wird auf Knoten 4.3.2 ausgeführt. Kann, Promise.allum es noch einmal zu wiederholen, so verwaltet werden, dass zu einem bestimmten Zeitpunkt nur eine bestimmte Anzahl von Versprechungen ausgeführt wird?

Beachten Sie, dass Promise.all()dies nicht die Versprechen auslöst, ihre Arbeit zu beginnen , sondern das Versprechen selbst erstellt.

Vor diesem Hintergrund besteht eine Lösung darin, zu prüfen, wann immer ein Versprechen gelöst wird, ob ein neues Versprechen gestartet werden soll oder ob Sie bereits am Limit sind.

Es ist jedoch wirklich nicht nötig, das Rad hier neu zu erfinden. Eine Bibliothek, die Sie für diesen Zweck verwenden könnten, istes6-promise-pool . Aus ihren Beispielen:

// 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)
})

P-Limit

Ich habe die Begrenzung der Versprechen-Parallelität mit einem benutzerdefinierten Skript, Bluebird, es6-Versprechen-Pool und p-Limit verglichen. Ich glaube, dass p-limit die einfachste, abgespeckte Implementierung für diesen Bedarf hat. Siehe ihre Dokumentation .

Bedarf

Zum Beispiel mit Async kompatibel sein

• ECMAScript 2017 (Version 8)
• Knotenversion > 8.2.1

Mein Beispiel

In diesem Beispiel müssen wir für jede URL im Array eine Funktion ausführen (z. B. eine API-Anforderung). Hier heißt das fetchData(). Wenn wir eine Reihe von Tausenden von Elementen verarbeiten müssten, wäre Parallelität definitiv nützlich, um CPU- und Speicherressourcen zu sparen.

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);
})();

Das Konsolenprotokollergebnis ist ein Array Ihrer Antwortdaten für aufgelöste Versprechen.

Wenn Sie wissen, wie Iteratoren funktionieren und wie sie verwendet werden, benötigen Sie keine zusätzliche Bibliothek, da es sehr einfach werden kann, Ihre eigene Parallelität selbst zu erstellen. Lassen Sie mich demonstrieren:

/* [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)
  }
}

Wir können denselben Iterator verwenden und ihn für alle Mitarbeiter freigeben.

Wenn Sie .entries()stattdessen verwendet .values()hätten, hätten Sie ein 2D-Array erstellt, mit [[index, value]]dem ich unten mit einer Parallelität von 2 demonstrieren werde

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

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'))

Dies hat den Vorteil, dass Sie eine Generatorfunktion haben können, anstatt alles auf einmal bereit zu haben.

Hinweis: Der Unterschied zum Beispiel- Async-Pool besteht darin, dass zwei Worker erzeugt werden. Wenn also ein Worker aus irgendeinem Grund einen Fehler bei Index 5 ausgibt, wird der andere Worker nicht davon abgehalten, den Rest zu erledigen. Sie gehen also von 2 Parallelität auf 1 zurück (damit es hier nicht aufhört). Mein Rat ist also, dass Sie alle Fehler innerhalb der doWorkFunktion abfangen

Verwenden von Array.prototype.splice

while (funcs.length) {
  // 100 at at time
  await Promise.all( funcs.splice(0, 100).map(f => f()) )
}

Die Promise.map von bluebird kann eine Parallelitätsoption verwenden, um zu steuern, wie viele Versprechen parallel ausgeführt werden sollen. Manchmal ist es einfacher, als .allweil Sie das Versprechen-Array nicht erstellen müssen.

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
}

Verwenden Sie anstelle von Versprechungen zur Begrenzung von http-Anforderungen die integrierten http.Agent.maxSockets des Knotens . Dies beseitigt die Notwendigkeit, eine Bibliothek zu verwenden oder Ihren eigenen Pooling-Code zu schreiben, und bietet den zusätzlichen Vorteil, dass Sie mehr Kontrolle darüber haben, was Sie einschränken.

agent.maxSockets

Standardmäßig auf Unendlich eingestellt. Legt fest, wie viele gleichzeitige Sockets der Agent pro Ursprung geöffnet haben kann. Origin ist entweder eine Kombination aus 'Host: Port' oder 'Host: Port: LocalAddress'.

Beispielsweise:

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

Wenn Sie mehrere Anforderungen an denselben Ursprung stellen, kann es auch von Vorteil sein, den Wert keepAliveauf true zu setzen (weitere Informationen finden Sie in den obigen Dokumenten).

Ich schlage den Async-Pool der Bibliothek vor: https://github.com/rxaviers/async-pool

npm install tiny-async-pool

Beschreibung:

Führen Sie mit nativem ES6 / ES7 mehrere Funktionen zur Rückgabe von Versprechungen und zur asynchronen Ausführung mit eingeschränkter Parallelität aus

asyncPool führt mehrere vielversprechende und asynchrone Funktionen in einem begrenzten Parallelitätspool aus. Es wird sofort abgelehnt, sobald eines der Versprechen abgelehnt wird. Es wird gelöst, wenn alle Versprechen erfüllt sind. Es ruft die Iteratorfunktion so schnell wie möglich auf (unter Parallelitätslimit).

Verwendung:

const timeout = i => new Promise(resolve => setTimeout(() => resolve(i), i));
await asyncPool(2, [1000, 5000, 3000, 2000], timeout);
// Call iterator (i = 1000)
// Call iterator (i = 5000)
// Pool limit of 2 reached, wait for the quicker one to complete...
// 1000 finishes
// Call iterator (i = 3000)
// Pool limit of 2 reached, wait for the quicker one to complete...
// 3000 finishes
// Call iterator (i = 2000)
// Itaration is complete, wait until running ones complete...
// 5000 finishes
// 2000 finishes
// Resolves, results are passed in given array order `[1000, 5000, 3000, 2000]`.

Hier ein grundlegendes Beispiel für Streaming und 'p-Limit'. Es überträgt http read stream an mongo db.

const stream = require('stream');
const util = require('util');
const pLimit = require('p-limit');
const es = require('event-stream');
const streamToMongoDB = require('stream-to-mongo-db').streamToMongoDB;


const pipeline = util.promisify(stream.pipeline)

const outputDBConfig = {
    dbURL: 'yr-db-url',
    collection: 'some-collection'
};
const limit = pLimit(3);

async yrAsyncStreamingFunction(readStream) => {
        const mongoWriteStream = streamToMongoDB(outputDBConfig);
        const mapperStream = es.map((data, done) => {
                let someDataPromise = limit(() => yr_async_call_to_somewhere())

                    someDataPromise.then(
                        function handleResolve(someData) {

                            data.someData = someData;    
                            done(null, data);
                        },
                        function handleError(error) {
                            done(error)
                        }
                    );
                })

            await pipeline(
                readStream,
                JSONStream.parse('*'),
                mapperStream,
                mongoWriteStream
            );
        }

Es kann durch Rekursion aufgelöst werden.

Die Idee ist, dass Sie anfangs die maximal zulässige Anzahl von Anforderungen senden und jede dieser Anforderungen sich nach Abschluss rekursiv selbst weiter senden sollte.

function batchFetch(urls, concurrentRequestsLimit) {
    return new Promise(resolve => {
        var documents = [];
        var index = 0;

        function recursiveFetch() {
            if (index === urls.length) {
                return;
            }
            fetch(urls[index++]).then(r => {
                documents.push(r.text());
                if (documents.length === urls.length) {
                    resolve(documents);
                } else {
                    recursiveFetch();
                }
            });
        }

        for (var i = 0; i < concurrentRequestsLimit; i++) {
            recursiveFetch();
        }
    });
}

var sources = [
    'http://www.example_1.com/',
    'http://www.example_2.com/',
    'http://www.example_3.com/',
    ...
    'http://www.example_100.com/'
];
batchFetch(sources, 5).then(documents => {
   console.log(documents);
});

Hier ist meine ES7-Lösung für eine Copy-Paste-freundliche und vollständige Promise.all()/ map()alternative Funktion mit einem Parallelitätslimit.

Ähnlich Promise.all()wird die Rückgabereihenfolge sowie ein Fallback für nicht versprochene Rückgabewerte beibehalten.

Ich habe auch einen Vergleich der verschiedenen Implementierungen beigefügt, da er einige Aspekte veranschaulicht, die einige der anderen Lösungen übersehen haben.

Verwendung

const asyncFn = delay => new Promise(resolve => setTimeout(() => resolve(), delay));
const args = [30, 20, 15, 10];
await asyncPool(args, arg => asyncFn(arg), 4); // concurrency limit of 4

Implementierung

async function asyncBatch(args, fn, limit = 8) {
  // Copy arguments to avoid side effects
  args = [...args];
  const outs = [];
  while (args.length) {
    const batch = args.splice(0, limit);
    const out = await Promise.all(batch.map(fn));
    outs.push(...out);
  }
  return outs;
}

async function asyncPool(args, fn, limit = 8) {
  return new Promise((resolve) => {
    // Copy arguments to avoid side effect, reverse queue as
    // pop is faster than shift
    const argQueue = [...args].reverse();
    let count = 0;
    const outs = [];
    const pollNext = () => {
      if (argQueue.length === 0 && count === 0) {
        resolve(outs);
      } else {
        while (count < limit && argQueue.length) {
          const index = args.length - argQueue.length;
          const arg = argQueue.pop();
          count += 1;
          const out = fn(arg);
          const processOut = (out, index) => {
            outs[index] = out;
            count -= 1;
            pollNext();
          };
          if (typeof out === 'object' && out.then) {
            out.then(out => processOut(out, index));
          } else {
            processOut(out, index);
          }
        }
      }
    };
    pollNext();
  });
}

Vergleich

// A simple async function that returns after the given delay
// and prints its value to allow us to determine the response order
const asyncFn = delay => new Promise(resolve => setTimeout(() => {
  console.log(delay);
  resolve(delay);
}, delay));

// List of arguments to the asyncFn function
const args = [30, 20, 15, 10];

// As a comparison of the different implementations, a low concurrency
// limit of 2 is used in order to highlight the performance differences.
// If a limit greater than or equal to args.length is used the results
// would be identical.

// Vanilla Promise.all/map combo
const out1 = await Promise.all(args.map(arg => asyncFn(arg)));
// prints: 10, 15, 20, 30
// total time: 30ms

// Pooled implementation
const out2 = await asyncPool(args, arg => asyncFn(arg), 2);
// prints: 20, 30, 15, 10
// total time: 40ms

// Batched implementation
const out3 = await asyncBatch(args, arg => asyncFn(arg), 2);
// prints: 20, 30, 20, 30
// total time: 45ms

console.log(out1, out2, out3); // prints: [30, 20, 15, 10] x 3

// Conclusion: Execution order and performance is different,
// but return order is still identical

Fazit

asyncPool() sollte die beste Lösung sein, da neue Anforderungen gestartet werden können, sobald eine vorherige abgeschlossen ist.

asyncBatch() wird als Vergleich aufgenommen, da die Implementierung einfacher zu verstehen ist, die Leistung jedoch langsamer sein sollte, da alle Anforderungen im selben Stapel abgeschlossen sein müssen, um den nächsten Stapel zu starten.

In diesem erfundenen Beispiel ist die nicht begrenzte Vanille Promise.all()natürlich die schnellste, während die anderen in einem realen Überlastungsszenario eine wünschenswertere Leistung erbringen könnten.

Aktualisieren

Die Async-Pool-Bibliothek, die andere bereits vorgeschlagen haben, ist wahrscheinlich eine bessere Alternative zu meiner Implementierung, da sie fast identisch funktioniert und eine präzisere Implementierung mit einer cleveren Verwendung von Promise.race () aufweist: https://github.com/rxaviers/ async-pool / blob / master / lib / es7.js

Hoffentlich kann meine Antwort immer noch einen pädagogischen Wert haben.

Also habe ich versucht, einige Beispiele für meinen Code zum Laufen zu bringen, aber da dies nur für ein Importskript und nicht für Produktionscode war, war die Verwendung der Stapelversprechen des npm-Pakets sicherlich der einfachste Weg für mich

HINWEIS: Benötigt Laufzeit, um Promise zu unterstützen oder um Polyfilled zu sein.

Api batchPromises (int: batchSize, array: Collection, i => Promise: Iteratee) Das Promise: Iteratee wird nach jedem Batch aufgerufen.

Verwenden:

batch-promises
Easily batch promises

NOTE: Requires runtime to support Promise or to be polyfilled.

Api
batchPromises(int: batchSize, array: Collection, i => Promise: Iteratee)
The Promise: Iteratee will be called after each batch.

Use:
import batchPromises from 'batch-promises';
 
batchPromises(2, [1,2,3,4,5], i => new Promise((resolve, reject) => {
 
  // The iteratee will fire after each batch resulting in the following behaviour:
  // @ 100ms resolve items 1 and 2 (first batch of 2)
  // @ 200ms resolve items 3 and 4 (second batch of 2)
  // @ 300ms resolve remaining item 5 (last remaining batch)
  setTimeout(() => {
    resolve(i);
  }, 100);
}))
.then(results => {
  console.log(results); // [1,2,3,4,5]
});

Rekursion ist die Antwort, wenn Sie keine externen Bibliotheken verwenden möchten

downloadAll(someArrayWithData){
  var self = this;

  var tracker = function(next){
    return self.someExpensiveRequest(someArrayWithData[next])
    .then(function(){
      next++;//This updates the next in the tracker function parameter
      if(next < someArrayWithData.length){//Did I finish processing all my data?
        return tracker(next);//Go to the next promise
      }
    });
  }

  return tracker(0); 
}

Dies habe ich Promise.racein meinem Code hier verwendet

const identifyTransactions = async function() {
  let promises = []
  let concurrency = 0
  for (let tx of this.transactions) {
    if (concurrency > 4)
      await Promise.race(promises).then(r => { promises = []; concurrency = 0 })
    promises.push(tx.identifyTransaction())
    concurrency++
  }
  if (promises.length > 0)
    await Promise.race(promises) //resolve the rest
}

Wenn Sie ein Beispiel sehen möchten : https://jsfiddle.net/thecodermarcelo/av2tp83o/5/

• @tcoocDie Antwort von war ziemlich cool. Wusste nichts davon und wird es in Zukunft nutzen.
• Ich habe auch die Antwort von @MatthewRideout genossen , aber es wird eine externe Bibliothek verwendet !!

Wann immer möglich, versuche ich, solche Dinge selbst zu entwickeln, anstatt in eine Bibliothek zu gehen. Am Ende lernst du viele Konzepte, die vorher entmutigend wirkten.

Was haltet ihr von diesem Versuch:
(Ich habe viel darüber nachgedacht und ich denke, es funktioniert, aber weist darauf hin, ob es nicht so ist oder ob etwas grundlegend falsch ist.)

 class Pool{
        constructor(maxAsync) {
            this.maxAsync = maxAsync;
            this.asyncOperationsQueue = [];
            this.currentAsyncOperations = 0
        }

        runAnother() {
            if (this.asyncOperationsQueue.length > 0 && this.currentAsyncOperations < this.maxAsync) {
                this.currentAsyncOperations += 1;
                this.asyncOperationsQueue.pop()()
                    .then(() => { this.currentAsyncOperations -= 1; this.runAnother() }, () => { this.currentAsyncOperations -= 1; this.runAnother() })
            }
        }

        add(f){  // the argument f is a function of signature () => Promise
            this.runAnother();
            return new Promise((resolve, reject) => {
                this.asyncOperationsQueue.push(
                    () => f().then(resolve).catch(reject)
                )
            })
        }
    }

//#######################################################
//                        TESTS
//#######################################################

function dbCall(id, timeout, fail) {
    return new Promise((resolve, reject) => {
        setTimeout(() => {
            if (fail) {
               reject(`Error for id ${id}`);
            } else {
                resolve(id);
            }
        }, timeout)
    }
    )
}


const dbQuery1 = () => dbCall(1, 5000, false);
const dbQuery2 = () => dbCall(2, 5000, false);
const dbQuery3 = () => dbCall(3, 5000, false);
const dbQuery4 = () => dbCall(4, 5000, true);
const dbQuery5 = () => dbCall(5, 5000, false);


const cappedPool = new Pool(2);

const dbQuery1Res = cappedPool.add(dbQuery1).catch(i => i).then(i => console.log(`Resolved: ${i}`))
const dbQuery2Res = cappedPool.add(dbQuery2).catch(i => i).then(i => console.log(`Resolved: ${i}`))
const dbQuery3Res = cappedPool.add(dbQuery3).catch(i => i).then(i => console.log(`Resolved: ${i}`))
const dbQuery4Res = cappedPool.add(dbQuery4).catch(i => i).then(i => console.log(`Resolved: ${i}`))
const dbQuery5Res = cappedPool.add(dbQuery5).catch(i => i).then(i => console.log(`Resolved: ${i}`))

Dieser Ansatz bietet eine nette API, ähnlich wie Thread-Pools in Scala / Java.
Nach dem Erstellen einer Instanz des Pools mitconst cappedPool = new Pool(2) , geben Sie ihm einfach Versprechen cappedPool.add(() => myPromise).
Ohne es zu merken, müssen wir sicherstellen, dass das Versprechen nicht sofort beginnt, und deshalb müssen wir es mit Hilfe der Funktion "träge" bereitstellen.

Beachten Sie vor allem, dass das Ergebnis der Methode add ein Versprechen ist, das mit dem Wert Ihres ursprünglichen Versprechens abgeschlossen / aufgelöst wird ! Dies ermöglicht eine sehr intuitive Bedienung.

const resultPromise = cappedPool.add( () => dbCall(...))
resultPromise
.then( actualResult => {
   // Do something with the result form the DB
  }
)

Leider gibt es keine Möglichkeit, dies mit nativem Promise.all zu tun. Sie müssen also kreativ sein.

Dies ist der schnellste und prägnanteste Weg, den ich finden konnte, ohne externe Bibliotheken zu verwenden.

Es verwendet eine neuere Javascript-Funktion, die als Iterator bezeichnet wird. Der Iterator verfolgt grundsätzlich, welche Elemente verarbeitet wurden und welche nicht.

Um es im Code zu verwenden, erstellen Sie ein Array von asynchronen Funktionen. Jede asynchrone Funktion fragt denselben Iterator nach dem nächsten Element, das verarbeitet werden muss. Jede Funktion verarbeitet ihr eigenes Element asynchron und fordert den Iterator nach Abschluss eines neuen Elements auf. Sobald dem Iterator die Elemente ausgehen, sind alle Funktionen abgeschlossen.

Vielen Dank an @Endless für die Inspiration.

var items = [
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
    "https://www.stackoverflow.com",
];

var concurrency = 5

Array(concurrency).fill(items.entries()).map(async (cursor) => {
    for(let [index, url] of cursor){
        console.log("getting url is ", index, url);
        // run your async task instead of this next line
        var text = await fetch(url).then(res => res.text());
        console.log("text is", text.slice(0,20));
    }
})