连接数百个RxScala Observable(每个都有数百万个要发出的事件)的有效方法？

Question

我有数据存储在磁盘上,数百万条记录的文件,每天一个.我有一个相对有效的反序列化器,可以生成发出记录的Observable,现在足够快(1.5M记录/秒).

我想要的是现在连接这些Observable,以便我可以得到一个多天的跨度不间断的流.当这个简单的工作时我很激动:

val nilObs: Observable[Record] = Observable.empty
val allObs = dates.map(reader.recordsForDate(_)).foldLeft(nilObs)(_ ++ _)

但是,上面的结果是Observable吞吐量很大 - 有200个连续的Observable,我看到50-100k/s,相比我预期的1.5M/s.

我还没有介绍过它,但是看看https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/internal/operators/OperatorConcat.java它正在做一个大量的队列工作 - 我想知道我是否在一个队列中创建了N个队列而不是N个Observable？

是否有更有效的方式以这种方式连接Observables？

Answer 1

肯定至少有一种更快的方法可以做到这一点:

val datesObs = Observable.from(dates.toIterable)
val allObs = datesObs.concatMap(reader.recordsForDate(_))

这有预期的表现,所以一切都很好.