为什么多个RXJava可观察对象没有并行执行？

Question

我是RxJava的新手，正在尝试从link执行多个观测值的并行执行示例：RxJava并行获取观测值

虽然上面链接中提供的示例是并行执行可观察对象，但是当我在foreach方法中添加一个Thread.sleep（TIME_IN_MILLISECONDS）时，系统开始一次执行一个可观察对象。请帮助我理解为什么Thread.sleep停止可观察对象的并行执行。

下面是导致多个观测值同步执行的修改示例：

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class ParallelExecution {

    public static void main(String[] args) {
        System.out.println("------------ mergingAsync");
        mergingAsync();
    }

    private static void mergingAsync() {
        Observable.merge(getDataAsync(1), getDataAsync(2)).toBlocking()
        .forEach(x -> { try{Thread.sleep(4000);}catch(Exception ex){}; 
        System.out.println(x + " " + Thread.currentThread().getId());});
    }

    // artificial representations of IO work
    static Observable<Integer> getDataAsync(int i) {
        return getDataSync(i).subscribeOn(Schedulers.io());
    }

    static Observable<Integer> getDataSync(int i) {
        return Observable.create((Subscriber<? super Integer> s) -> {
            // simulate latency
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
            s.onNext(i);
            s.onCompleted();
        });
    }
}

在上面的示例中，我们使用了observable的subscribeOn方法，并提供了一个线程池（Schedules.io）来执行，因此每个observable的订阅将在单独的线程上进行。

Answer 1

实际上，在您的示例中，并行执行正在发生，您只是看得不正确，在执行工作的位置和发出通知的位置之间存在差异。

如果你将日志与线程id在Observable.create，你会注意到每个可观察是在不同的线程同时执行。