Wie kann die Parallelität von flatMap eingeschränkt werden?

Question

Ich versuche RxJS zu verwenden, um ein Skript zu schreiben, um mehrere Hunderte von Protokolldateien zu verarbeiten, von denen jede ungefähr 1 GB ist. Das Skelett des Skripts sieht aus wie

Rx.Observable.from(arrayOfLogFilePath) 
.flatMap(function(logFilePath){ 
    return Rx.Node.fromReadStream(logFilePath) 
    .filter(filterLogLine) 
}) 
.groupBy(someGroupingFunc) 
.map(someFurtherProcessing) 
.subscribe(...) 

Der Code funktioniert, merkt aber, dass der Filterschritt aller Protokolldateien gleichzeitig starten. Aus Sicht der E/A-Performance des Dateisystems ist es jedoch vorzuziehen, eine Datei nach der anderen zu verarbeiten (oder zumindest die Nebenläufigkeit auf einige Dateien zu beschränken, anstatt alle hunderten von Dateien gleichzeitig zu öffnen). Wie kann ich es in diesem Sinne "funktional reaktiv" umsetzen?

Ich hatte an Scheduler gedacht, konnte aber nicht herausfinden, wie es hier helfen kann.

Answer 1

Sie können .merge(maxConcurrent) verwenden, um die Parallelität zu begrenzen. Weil .merge(maxConcurrent) eine Metaobservable (beobachtbar von Observablen) in eine Observable abflacht, müssen Sie die .flatMap durch .map ersetzen, so dass die Ausgabe eine Metaobservable ("unflat") ist, dann rufen Sie .merge(maxConcurrent).

Rx.Observable.from(arrayOfLogFilePath) 
.map(function(logFilePath){ 
    return Rx.Node.fromReadStream(logFilePath) 
    .filter(filterLogLine) 
}) 
.merge(2) // 2 concurrent 
.groupBy(someGroupingFunc) 
.map(someFurtherProcessing) 
.subscribe(...) 

Dieser Code wurde nicht getestet (da ich Sie haben keinen Zugriff auf die Umwelt Entwicklung haben), aber das ist, wie Sie vorgehen. RxJS hat nicht viele Operatoren mit Parallelitätsparametern, aber Sie können fast immer tun, was Sie mit .merge(maxConcurrent) benötigen.

Answer 2

Ich habe gerade ein ähnliches Problem mit RxJs 5 gelöst, also hoffe ich, dass die Lösung anderen mit einem ähnlichen Problem helfen kann.

// Simulate always processing 2 requests in parallel (when one is finished it starts processing one more), 
 
// retry two times, push error on stream if retry fails. 
 

 
//const Rx = require('rxjs-es6/Rx'); 
 

 
// -- Global variabel just to show that it works. -- 
 
let parallelRequests = 0; 
 
// -------------------------------------------------- 
 

 
function simulateRequest(req) { 
 
    console.log("Request " + req); 
 
    // --- To log retries --- 
 
    var retry = 0; 
 
    // ---------------------- 
 

 
    // Can't retry a promise, need to restart before the promise is made. 
 
    return Rx.Observable.of(req).flatMap(req => new Promise((resolve, reject) => { 
 

 
     var random = Math.floor(Math.random() * 2000); 
 
     // -- To show that it works -- 
 
     if (retry) { 
 
      console.log("Retrying request " + req + " ,retry " + retry); 
 
     } else { 
 

 
      parallelRequests++; 
 
     } 
 
     // --------------------------- 
 
     setTimeout(() => { 
 
      if (random < 900) { 
 
       retry++; 
 
       return reject(req + " !!!FAILED!!!"); 
 
      } 
 

 
      return resolve(req); 
 
     }, random); 
 
    })).retry(2).catch(e => Rx.Observable.of(e)); 
 
} 
 

 
Rx.Observable.range(1, 10) 
 
    .flatMap(e => simulateRequest(e), null, 2) 
 
    // -- To show that it works -- 
 
    .do(() => { 
 
     console.log("ParallelRequests " + parallelRequests); 
 
     parallelRequests--; 
 
    }) 
 
    // --------------------------- 
 
    .subscribe(e => console.log("Response from request " + e), e => console.log("Should not happen, error: " + e), e => console.log("Finished"));

<script src="https://npmcdn.com/@reactivex/rxjs@5.0.0-beta.6/dist/global/Rx.umd.js"></script>