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

发布时间：2020-12-16 10:04:57 所属栏目：安全来源：网络整理

导读：我有数据存储在磁盘上,数百万条记录的文件,每天一个.我有一个相对有效的反序列化器,可以生成发出记录的Observable,现在足够快(1.5M记录/秒). 我想要的是现在连接这些Observable,以便我可以得到一个多天的跨度不间断的流.当这个简单的工作时我很激动： val ni

我有数据存储在磁盘上,数百万条记录的文件,每天一个.我有一个相对有效的反序列化器,可以生成发出记录的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？

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

解决方法

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

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

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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!