RxJava2 Flowable zip 异步方法调用答案

【问题标题】：RxJava2 Flowable zip async method invokeRxJava2 Flowable zip 异步方法调用
【发布时间】：2018-03-15 22:02:56
【问题描述】：

一个简单的例子：

import io.reactivex.Flowable;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RxTest {

    private static Logger log = LoggerFactory.getLogger(RxTest.class);

    public static void main(String[] args) throws InterruptedException     {
        log.debug("start");
        RxTest rx = new RxTest();
        Flowable.zip(rx.m1(), rx.m2(), rx::result).subscribe( (r) -> log.debug(r));
    }

    private Publisher<String> m1() throws InterruptedException {
        log.debug("m2");
        String result = new TestRxObj().getObj(4000L, "Hello");
        return Flowable.just(result);
    }

    private Publisher<Long> m2() throws InterruptedException {
        log.debug("m1");
        Long result = new TestRxObj().getObj(100L, 777L);
        return Flowable.just(result);
    }

    private String result(String t1, Long t2) {
        log.debug("result");
        return t1 + "-" + t2;
    }

}

public class TestRxObj {

    public <T> T getObj(Long time, T t) throws InterruptedException {
        Thread.sleep(time);
        return t;
    }

}

所以问题是，我想异步调用方法 m1() 和 m2()

现在我有这个输出：

2018-03-09 22:34:20 [main] DEBUG RxTest: 18 - start
2018-03-09 22:34:20 [main] DEBUG RxTest: 24 - m2
2018-03-09 22:34:24 [main] DEBUG RxTest: 30 - m1
2018-03-09 22:34:24 [main] DEBUG RxTest: 36 - result
2018-03-09 22:34:24 [main] DEBUG RxTest: 20 - Hello-777

在 git 上，他们有一些例子，如何在新线程中运行：

import io.reactivex.schedulers.Schedulers;

Flowable.fromCallable(() -> {
    Thread.sleep(1000); //  imitate expensive computation
    return "Done";
})
  .subscribeOn(Schedulers.io())
  .observeOn(Schedulers.single())
  .subscribe(System.out::println, Throwable::printStackTrace);

Thread.sleep(2000); // <--- wait for the flow to finish

但是这个例子的问题，我不知道如何加入（主要/当前）威胁，不写Thread.sleep(2000);。我不是在 Android 上运行的，只是简单的 java 项目，所以我不能.observeOn(AndroidSchedulers.mainThread())，如果我正确理解这是做什么的

我的主要目标是使用 RxJava2 异步运行 1..N(至少 2 :)) 方法，如果它可以在没有zip 的情况下完成，我可以接受，任何使用 RxJava 的方法都没有Thread.sleep(...) :)

【问题讨论】：

我希望你意识到 m1() 在 RxJava 参与之前就被调用和执行了。返回 Java 主线程的唯一方法是阻塞订阅或使用阻塞调度程序。使用blockingSubscribe，前者更容易。

标签： java asynchronous rx-java rx-java2

【解决方案1】：

也许这会有所帮助：

public static void main(String[] args) throws InterruptedException     {
    log.debug("start");
    RxTest rx = new RxTest();
    Flowable.zip(
        Observable.defer(() -> rx.m1()).subscribeOn(Schedulers.io()),
        Observable.defer(() -> rx.m2()).subscribeOn(Schedulers.io()),
        rx::result)
    .subscribe( (r) -> log.debug(r));
}

出于测试目的，有时将Schedulers.io() 作为参数或字段很有用，因此您可以将其替换为TestScheduler。

【讨论】：