Come limitare la concorrenza di flatMap?

Question

Sto cercando di utilizzare RxJS per scrivere uno script per elaborare diverse centinaia di file di registro, ognuno dei quali è di circa 1 GB. Lo scheletro dello script appare come

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

Il codice funziona, a meno di notare che la fase di filtraggio di tutti i file di registro avrà inizio contemporaneamente. Tuttavia, dal punto di vista delle prestazioni IO del file system, è preferibile elaborare un file dopo l'altro (o almeno limitare la concorrenza a pochi file anziché aprire tutte le centinaia di file nello stesso momento). A questo proposito, come posso implementarlo in un "modo funzionale reattivo"?

Avevo pensato all'utilità di pianificazione ma non riuscivo a capire come può essere d'aiuto.

Answer 1

È possibile utilizzare .merge(maxConcurrent) per limitare la concorrenza. Poiché .merge(maxConcurrent) appiattisce un osservabile (osservabile di osservabili) in un osservabile, è necessario sostituire lo .flatMap con .map in modo che l'output sia un metaosservabile ("unflat"), quindi si chiama .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(...) 

Questo codice non è stato testato (dal momento che non hanno accesso all'ambiente di sviluppo di avere), ma questo è il modo di procedere. RxJS non ha molti operatori con parametri di concorrenza, ma puoi fare quasi sempre ciò che ti serve con .merge(maxConcurrent).

Answer 2

Ho appena risolto un problema simile con RxJs 5, quindi spero che la soluzione possa aiutare gli altri con un problema simile.

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