Batch-Modus mit Versprechen in Javascript

Question

Ich muss eine Menge Daten verarbeiten. Für jeden Eintrag der Daten müssen wir eine Anfrage an mysql senden. Meine aktuelle Lösung ist wie folgt:

var Q = require('q'); 
    function process(entry){ 
     ... 
     var defered=Q.defer(); 
     connection.query(sql,defered.makeNodeResolver()); 
     return defered.promise; 
    } 
    function ProcessAll(results) { 
     var waitfor=[]; 
     for(var i=0;i< results.length;i++){ 
      waitfor.push(process(results[i])); 
     } 
     Q.all(waitfor).then(function(results) { 
      notifySuc(results); 
     },function(results){ 
      notifyFail(results); 
     }); 
    } 

jedoch, wenn die Anzahl der Daten sehr groß ist, wird es durch Absturz aus dem Speicher zu:

FATAL ERROR: Committing semi space failed. Allocation failed - process out of memory 

1: node::Abort() [node] 
2: 0x109624c [node] 
3: v8::Utils::ReportApiFailure(char const*, char const*) [node] 
4: v8::internal::V8::FatalProcessOutOfMemory(char const*, bool) [node] 
5: v8::internal::Heap::PerformGarbageCollection(v8::internal::GarbageCollector, v8::GCCallbackFlags) [node] 
6: v8::internal::Heap::CollectGarbage(v8::internal::GarbageCollector, char const*, char const*, v8::GCCallbackFlags) [node] 

Wie kann ich es mit kleineren Chargen zu brechen? Zum Beispiel können wir 1000 Einträge jedes Mal verarbeiten. Nachdem sie alle eins sind, nehmen wir den Rest wieder auf. Wie macht man das mit den Callback-Funktionen?

Können wir außerdem mehrere Einträge parallel verarbeiten lassen?

Answer 1

Sie könnten etwas wie BaconJS verwenden, um dies zu tun. Sie können einen Ereignisdatenstrom mit Bacon.fromPromise (provide [, abort] [, eventTransformer]) erstellen und dann stream.bufferWithCount (Anzahl) verwenden, um die Verarbeitung in Stapel aufzubrechen. BaconJS ist eine kleine Bibliothek für funktionale reaktive Programmierung. Um mehr über FRP zu erfahren, können Sie die The Introduction to FRP für eine schnelle Grundierung lesen.

Answer 2

Wenn der Fehler nicht aufgrund notifySuc ist man leicht eine Charge zu einer Zeit verarbeiten kann wie folgt

var Q = require('q'); 
function process(entry){ 
    ... 
    var defered=Q.defer(); 
    connection.query(sql,defered.makeNodeResolver()); 
    return defered.promise; 
} 
function processBatch(batch) { 
    return Q.all(batch.map(item => process(item))); 
} 
function ProcessAll(results) { 
    var batchSize = 1000; 
    var pos = 0; 
    var promises = Q([]); // initial promise of empty array 
    while (pos < results.length) { 
     (function(pos) { 
      promises = promises.then(result => processBatch(results.slice(pos, batchSize)).then(results => result.concat(results))); 
     })(pos); 
     pos += batchSize; 
    } 
    promises.then(function(results) { 
     notifySuc(results); 
    },function(results){ 
     notifyFail(results); 
    }); 
}