Batch anfragen in Node.js

Question

Mein Programm kommuniziert mit einem Webdienst, der nur ~10 Anfragen pro Sekunde akzeptiert. Von Zeit zu Zeit sendet mein Programm über 100 gleichzeitige Anfragen an den Webdienst, was dazu führt, dass mein Programm abstürzt.

Wie kann ich gleichzeitige Anfragen in Node.js auf 5 pro Sekunde begrenzen? Ich verwende die Anfragembibliothek.

 // IF EVENT AND SENDER
    if(data.sender[0].events && data.sender[0].events.length > 0) {

        // FIND ALL EVENTS
        for(var i = 0; i < data.sender[0].events.length; i++) {

            // IF TYPE IS "ADDED"
            if(data.sender[0].events[i].type == "added") {
                switch (data.sender[0].events[i].link.rel) {
                    case "contact" :
                        batch("added", data.sender[0].events[i].link.href);
                        //_initContacts(data.sender[0].events[i].link.href);
                        break;
                } 
            // IF TYPE IS "UPDATED"
            } else if(data.sender[0].events[i].type == "updated") {

                switch (data.sender[0].events[i].link.rel){                     
                    case "contactPresence" :
                        batch("updated", data.sender[0].events[i].link.href);
                        //_getContactPresence(data.sender[0].events[i].link.href);
                        break;
                    case "contactNote" :
                        batch("updated", data.sender[0].events[i].link.href);
                        // _getContactNote(data.sender[0].events[i].link.href);
                        break;
                    case "contactLocation" :
                        batch("updated", data.sender[0].events[i].link.href);
                        // _getContactLocation(data.sender[0].events[i].link.href);
                        break;
                    case "presenceSubscription" :
                        batch("updated", data.sender[0].events[i].link.href);
                        // _extendPresenceSubscription(data.sender[0].events[i].link.href);
                        break;
                }
            }
        };

Und dann die selbstgemachte Batch-Methode:

var updated = [];
var added = [];

var batch = function(type, url){
    console.log("Batch aufgerufen");

    if (type === "added"){
        console.log("Hinzufügen ins Batch");
        added.push(url);
        if (added.length > 5) {
            setTimeout(added.forEach(function(req){
                _initContacts(req);
            }), 2000);
            added = [];
        }
    } 
    else if (type === "updated"){
        console.log("Aktualisierung ins Batch");
        updated.push(url);
        console.log("Aktualisierungslänge ist : ", updated.length);
        if (updated.length > 5){
            console.log("Mehr als 5 aktualisierte Ereignisse");
            updated.forEach(function(req){
                setTimeout(_getContactLocation(req), 2000);
            });
            updated = [];
        }
    }       
};

Und ein Beispiel für die tatsächliche Anfrage:

var _getContactLocation = function(url){
    r.get(baseUrl + url, 
    { "strictSSL" : false, "headers" : { "Authorization" : "Bearer " + accessToken }}, 
        function(err, res, body){
            if(err)
                console.log(err);
            else {
                var data = JSON.parse(body);
                self.emit("data.contact", data);
            }
        }
    );
};

Answer 1

Mit der async-Bibliothek führt die mapLimit-Funktion genau das aus, was du möchtest. Ich kann kein Beispiel für deinen spezifischen Anwendungsfall bereitstellen, da du keinen Code bereitgestellt hast.

Aus der Readme:

---

mapLimit(arr, limit, iterator, callback)

Das gleiche wie map, nur dürfen nicht mehr als "limit" Iteratoren gleichzeitig ausgeführt werden.

Beachte, dass die Elemente nicht in Batches verarbeitet werden, daher besteht keine Garantie dafür, dass die ersten "limit" Iterator-Funktionen abschließen, bevor andere gestartet werden.

Argumente

• arr - Ein Array, über das iteriert werden soll.
• limit - Die maximale Anzahl von Iteratoren, die gleichzeitig ausgeführt werden sollen.
• iterator(item, callback) - Eine Funktion, die auf jedes Element im Array angewendet wird. Dem Iterator wird ein callback(err, transformed) übergeben, der aufgerufen werden muss, sobald die Verarbeitung abgeschlossen ist, mit einem Fehler (der auch null sein kann) und einem transformierten Element.
• callback(err, results) - Ein Rückruf, der aufgerufen wird, nachdem alle Iteratorfunktionen abgeschlossen sind oder ein Fehler aufgetreten ist. Results ist ein Array der transformierten Elemente aus dem ursprünglichen Array.

Beispiel

async.mapLimit(['datei1','datei2','datei3'], 1, fs.stat, function(err, results){ // results ist jetzt ein Array von Statistiken für jede Datei });

---

EDIT: Jetzt, da du Code bereitgestellt hast, sehe ich, dass dein Einsatz etwas anders ist als ich angenommen habe. Die async-Bibliothek ist nützlicher, wenn du im Voraus alle Aufgaben kennst. Mir ist kein Bibliothek bekannt, die dieses Problem leicht für dich lösen kann. Die obige Anmerkung ist wahrscheinlich immer noch relevant für Personen, die nach diesem Thema suchen, also lasse ich sie stehen.

Es tut mir leid, ich habe keine Zeit, deinen Code umzustrukturieren, aber dies ist ein (nicht getestetes) Beispiel für eine Funktion, die eine asynchrone Anfrage macht und sich gleichzeitig auf 5 Anfragen pro Sekunde selbst drosselt. Ich empfehle dringend, darauf aufzubauen, um eine allgemeinere Lösung zu finden, die zu deiner Code-Basis passt.

var throttledRequest = (function () {
    var queue = [], running = 0;

    function sendPossibleRequests() {
        var url;
        while (queue.length > 0 && running < 5) {
            url = queue.shift();
            running++;
            r.get(url, { /* DEINE OPTIONEN HIER*/ }, function (err, res, body) {
                running--;
                sendPossibleRequests();

                if(err)
                    console.log(err);
                else {
                    var data = JSON.parse(body);
                    self.emit("data.contact", data);
                }
            });
        }
    }

    return function (url) {
        queue.push(url);
        sendPossibleRequests();
    };
})();

Im Wesentlichen behältst du eine Warteschlange aller Daten, die asynchron verarbeitet werden sollen (wie URLs, die angefordert werden sollen), und nach jedem Rückruf (von einer Anfrage) versuchst du, so viele verbleibende Anfragen wie möglich zu starten.

Answer 2

Dies ist genau das, worum es in der Agent Klasse von Node geht. Haben Sie etwas Dummes getan wie require('http').globalAgent.maxSockets = Number.MAX_VALUE oder agent: false als Anforderungsoption übergeben?

Mit dem Standardverhalten von Node sendet Ihr Programm nicht mehr als 5 gleichzeitige Anfragen auf einmal. Darüber hinaus bietet der Agent Optimierungen, die eine einfache Warteschlange nicht kann (insbesondere HTTP-Keepalives).

Wenn Sie viele Anfragen stellen (zum Beispiel 100 Anfragen aus einer Schleife herausgeben), werden die ersten 5 beginnen und der Agent wird die restlichen 95 in die Warteschlange stellen. Wenn Anfragen abgeschlossen sind, startet er die nächsten.

Was Sie wahrscheinlich tun möchten, ist einen Agent für Ihre Web-Service-Anfragen zu erstellen und ihn bei jedem Aufruf von request zu übergeben (anstelle von Anfragen mit dem globalen Agent zu mischen).

var http=require('http'), svcAgent = http.Agent();

request({ ... , agent: svcAgent });