我正在尝试使用RxJS编写一个脚本来处理数百个日志文件,每个文件大约1GB.脚本的骨架看起来像
Rx.Observable.from(arrayOfLogFilePath)
.flatMap(function(logFilePath){
return Rx.Node.fromReadStream(logFilePath)
.filter(filterLogLine)
})
.groupBy(someGroupingFunc)
.map(someFurtherProcessing)
.subscribe(...)
代码有效,但请注意所有日志文件的过滤步骤将同时启动.但是,从文件系统IO性能的角度来看,最好一个接一个地处理一个文件(或者至少将并发限制为几个文件而不是同时打开所有数百个文件).在这方面,我如何以“功能反应方式”实施？
我曾想过调度程序,但无法弄清楚它在这里有什么用处.
解决方法
您可以使用
.merge(maxConcurrent)来限制并发性.因为.merge(maxConcurrent)将一个Metaobservable(可观察的observables)展平为一个observable,你需要用.map替换.flatMap以便输出是一个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)做你需要的.