我有一个Observable<String>
。我想将其转换为Map<String, Int>
,它告诉我每个不同字符串的出现次数。
可观察的元素包含约10亿个元素,其中有1000个是不同的(因此,不能将整个数据集存储在RAM中)。目前,我遍历Observable
并更新HashMap
。我还确保在同一线程上进行观察,以避免出现竞争情况。但是,获得元件频率本来应该易于并行化,因此最好使用它。
有没有办法做到这一点?
最佳答案
您可以使用groupBy
代替自己维护HashMap
。 groupBy
将为每个密钥创建一个Observable
,您可以在其他Scheduler上进行预订。例如。,
public class KeyCounter {
int key;
long count;
public KeyCounter(int key, long count) {
this.key = key;
this.count = count;
}
@Override
public String toString() {
return "key: " + key + " count: " + count;
}
}
@Test
public void foo() {
Observable<Integer> o = Observable.just(1, 2, 3, 2, 1);
o.groupBy(i -> i).flatMap(
group ->
group.subscribeOn(Schedulers.computation()).countLong().map(count -> new KeyCounter(group.getKey(), count))
).subscribe(System.out::println);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}