我是RxJava的新手,正在努力解决一个(我认为)简单的问题。我想在3个线程中同时处理订阅部分。那就是为什么我使用FixedThreadPool。示例代码:
Observer.just("one", "two", "three", "four")
.observeOn(Schedulers.io())
.subscribeOn(Schedulers.from(Executors.newFixedThreadPool(3))
.subscribe(new Observer<String>() {
public void onNext(String string) {
Log.d(TAG, "Started: " + string);
SystemClock.sleep(1000);
Log.d(TAG, "Ended: " + string);
}
(...)
}
预期结果:
Started: one
Started: two
Started: three
Ended: one
Started: four
Ended: two
Ended: three
Ended: four
实际结果:
Started: one
Ended: one
Started: two
Ended: two
Started: three
Ended: three
Started: four
Ended: four
我究竟做错了什么?
最佳答案
RxJava Observables是顺序的,并且subscribeOn
和observeOn
运算符不会并行运行值。
您可以实现的最接近的事情是通过模键对值进行分组,通过observeOn
运行它们并合并结果:
AtomicInteger count = new AtomicInteger();
Observable.range(1, 100)
.groupBy(v -> count.getAndIncrement() % 3)
.flatMap(g -> g
.observeOn(Schedulers.computation())
.map(v -> Thread.currentThread() + ": " + v))
.toBlocking()
.forEach(System.out::println);