可观测数据并行执行

Question

我用reactiveX Zip做了一些实验，我注意到我在zip中定义的可观察性是一个接一个地执行的。我认为zip的好处是，zip中定义的每一个可观察到的线程都是由一个线程执行的，所以所有这些线程都是并行执行的。有什么方法能达到我想要的吗？。这是我的zip例子

         @Test
public void testZip() {
    Observable.zip(obString(), obString1(), obString2(), (s, s2, s3) -> s.concat(s2)
                                                                     .concat(s3))
              .subscribe(System.out::println);
}

public Observable<String> obString() {
    System.out.println(Thread.currentThread().getId());
    return Observable.just("hello");
}

public Observable<String> obString1() {
    System.out.println(Thread.currentThread().getId());
    return Observable.just(" world");
}

public Observable<String> obString2() {
    System.out.println(Thread.currentThread().getId());
    return Observable.just("!");
}

Answer 1

你看错东西了。

obString*都在同一个线程上执行，因为它们是在testZip中调用它们时执行的。

您想要查看的是可观察的内容，这是不可能使用简单的只是，您需要一个自定义的可观察的内容，并查看onSubcribe主体中的当前线程。

此外，您可能希望使用scheduleOn为您的Observable提供一个特定的新线程或线程池。