如何限制flatMap的并发？答案

【问题标题】：How to limit the concurrency of flatMap?如何限制flatMap的并发？
【发布时间】：2014-09-30 00:44:21
【问题描述】：

我正在尝试使用 RxJS 编写一个脚本来处理数百个日志文件，每个日志文件大约 1GB。脚本的骨架看起来像

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

代码有效，但请注意所有日志文件的过滤步骤将同时开始。但是，从文件系统 IO 性能的角度来看，最好是一个接一个地处理一个文件（或者至少将并发限制在几个文件，而不是同时打开所有数百个文件）。在这方面，我如何以“功能反应方式”实现它？

我曾想过调度程序，但在这里无法弄清楚它如何提供帮助。

【问题讨论】：

我也有同样的问题，但使用的是 Rx.NET。可能吗？ stackoverflow.com/questions/37345516/…

标签： javascript rxjs

【解决方案1】：

您可以使用.merge(maxConcurrent) 来限制并发。因为.merge(maxConcurrent)将一个metaobservable（observable的observable）扁平化为一个observable，所以需要将.flatMap替换为.map，这样输出就是一个metaobservable（“unflat”），然后调用.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(...)

此代码尚未经过测试（因为我无权访问您拥有的开发环境），但这是继续进行的方法。 RxJS 没有很多带有并发参数的操作符，但你几乎总是可以用.merge(maxConcurrent) 做你需要的事情。

【讨论】：

这正是我想要开始工作的。我有一个要加载的 500 个 URL 的列表，并且不想同时启动所有请求。我使用了 map(5)，但它不起作用...所有请求都是同时发出的。
@Roaders 你让这个解决方案工作了吗？我也在尝试同样的方法。但是，所有请求都会同时触发。我用谷歌搜索了所有地方，但没有找到任何东西。
例如，如果您正在进行异步 http 调用，则需要将其包装在 Rx.defer() 中，以便 Rx 可以决定何时进行调用（例如，如果调用失败则重试）

【解决方案2】：

我刚刚用 RxJs 5 解决了一个类似的问题，所以我希望这个解决方案可以帮助其他有类似问题的人。

// 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"));

&lt;script src="https://npmcdn.com/@reactivex/rxjs@5.0.0-beta.6/dist/global/Rx.umd.js"&gt;&lt;/script&gt;

【讨论】：