如何在线程池队列上观察我的快速源

Question

需要帮助在主线程上进行可观察的启动，然后转到线程池以允许源继续发出新项目（无论它们是否仍在线程池中处理）。

这是我的例子：

public static void main(String[] args) {
    Observable<Integer> source = Observable.range(1,10);

    source.map(i -> sleep(i, 10))
            .doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))
            .observeOn(Schedulers.computation())
            .map(i -> sleep(i * 10, 300))
            .subscribe( i -> System.out.println("Received " + i + " on thread " + Thread.currentThread().getName()));

    sleep(-1, 30000);
}

private static int sleep(int i, int time) {
    try {
        Thread.sleep(time);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return i;
}

总是打印：

Emitting 1 on thread main
Emitting 2 on thread main
Emitting 3 on thread main
Received 10 on thread RxComputationScheduler-1
Emitting 4 on thread main
Emitting 5 on thread main
Emitting 6 on thread main
Received 20 on thread RxComputationScheduler-1
Emitting 7 on thread main
Emitting 8 on thread main
Emitting 9 on thread main
Received 30 on thread RxComputationScheduler-1
Emitting 10 on thread main
Received 40 on thread RxComputationScheduler-1
Received 50 on thread RxComputationScheduler-1
Received 60 on thread RxComputationScheduler-1
Received 70 on thread RxComputationScheduler-1
Received 80 on thread RxComputationScheduler-1
Received 90 on thread RxComputationScheduler-1
Received 100 on thread RxComputationScheduler-1

虽然项目按预期在主线程上发出，但我希望它们之后继续移动到计算/IO 线程池。

应该是这样的：

Answer 1

我认为您没有充分减缓源排放，而且它们的排放速度如此之快，以至于在observeOn() 有机会安排它们之前所有项目都已排放。

尝试睡眠到 500 毫秒而不是 10 毫秒。然后你会看到你所期望的交错。

public class JavaLauncher {
    public static void main(String[] args) {
        Observable<Integer> source = Observable.range(1,10);

        source.map(i -> sleep(i, 500))
                .doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))
                .observeOn(Schedulers.computation())
                .map(i -> sleep(i * 10, 250))
                .subscribe( i -> System.out.println("Received " + i + " on thread " + Thread.currentThread().getName()));

        sleep(-1, 30000);
    }

    private static int sleep(int i, int time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }
}

输出

Emitting 1 on thread main
Emitting 2 on thread main
Emitting 3 on thread main
Received 10 on thread RxComputationThreadPool-3
Emitting 4 on thread main
Received 20 on thread RxComputationThreadPool-3
Emitting 5 on thread main
Emitting 6 on thread main
Received 30 on thread RxComputationThreadPool-3
Emitting 7 on thread main
Emitting 8 on thread main
Received 40 on thread RxComputationThreadPool-3
Emitting 9 on thread main
Emitting 10 on thread main
Received 50 on thread RxComputationThreadPool-3
Received 60 on thread RxComputationThreadPool-3
Received 70 on thread RxComputationThreadPool-3
Received 80 on thread RxComputationThreadPool-3
Received 90 on thread RxComputationThreadPool-3
Received 100 on thread RxComputationThreadPool-3

更新 - 并行版本

public class JavaLauncher {
    public static void main(String[] args) {
        Observable<Integer> source = Observable.range(1,10);

        source.map(i -> sleep(i, 250))
                .doOnNext(i -> System.out.println("Emitting " + i + " on thread " + Thread.currentThread().getName()))
                .flatMap(i -> 
                    Observable.just(i)
                        .subscribeOn(Schedulers.computation())
                        .map(i2 -> sleep(i2 * 10, 500))
                )
                .subscribe( i -> System.out.println("Received " + i + " on thread " + Thread.currentThread().getName()));

        sleep(-1, 30000);
    }

    private static int sleep(int i, int time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return i;
    }
}

输出

Emitting 1 on thread main
Emitting 2 on thread main
Emitting 3 on thread main
Received 10 on thread RxComputationThreadPool-3
Emitting 4 on thread main
Received 20 on thread RxComputationThreadPool-4
Received 30 on thread RxComputationThreadPool-1
Emitting 5 on thread main
Received 40 on thread RxComputationThreadPool-2
Emitting 6 on thread main
Received 50 on thread RxComputationThreadPool-3
Emitting 7 on thread main
Received 60 on thread RxComputationThreadPool-4
Emitting 8 on thread main
Received 70 on thread RxComputationThreadPool-1
Emitting 9 on thread main
Received 80 on thread RxComputationThreadPool-2
Emitting 10 on thread main
Received 90 on thread RxComputationThreadPool-3
Received 100 on thread RxComputationThreadPool-4