如何在Android上使用RxJava在多个线程上运行subsciber

Question

我是RxJava的新手并且在(我猜)简单问题上苦苦挣扎.我想在3个线程中处理订阅部分simuleasly.这就是我使用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

我究竟做错了什么？

Answer 1

RxJava Observable是顺序的,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);