Rxjava Scheduler.trampoline与连接映射

Question

似乎基于文档，Scheduler.trampoline确保元素发出的是先入先出(即按顺序)。concat map的要点似乎也是为了确保所有内容都被适当地排列起来，然后发出。所以我想知道，应用subscribeOn./.observeOn(Scheduler.trampoline())，然后进行连接映射运算符，而不是常规的映射操作，是否有任何意义。

Answer 1

是的，这是有道理的。举个例子：

Observable.just(1, 2, 3, 4, 5)
    .subscribeOn(Schedulers.trampoline())
    .flatMap(
        a -> {
          if (a < 3) {
            return Observable.just(a).delay(3, TimeUnit.SECONDS);
          } else {
            return Observable.just(a);
          }
        })
    .doOnNext(
        a -> System.out.println("Element: " + a + ", on: " + Thread.currentThread().getName()))
    .subscribe();

下面是输出：

Element: 3, on: main
Element: 4, on: main
Element: 5, on: main
Element: 1, on: RxComputationScheduler-1
Element: 2, on: RxComputationScheduler-2

这里发生的情况是1和2依次到达flatMap运算符。但是现在，这些元素的内部流被延迟了3秒。请注意，flatMap急切地subscribes到内部流。也就是说，它不会在subscribing到下一个内部流(这就是concatMap所做的事情)之前等待一个流完成(使用onComplete)。

因此，1和2的内部流延迟了3秒。您可以说这是一个需要花费一些时间的外部I/O调用。同时，接下来的3个元素(3,4,5)进入flatMap，它们的流立即结束。这就是为什么您在输出中看到保持序列的原因。

然后3秒过去，元素1和2被发出。请注意，不能保证1会在2之前。

现在用concatMap替换flatMap，您将看到序列保持不变：

Element: 1, on: RxComputationScheduler-1
Element: 2, on: RxComputationScheduler-2
Element: 3, on: RxComputationScheduler-2
Element: 4, on: RxComputationScheduler-2
Element: 5, on: RxComputationScheduler-2

为什么？因为这就是concatMap的工作方式。元素1出现，并在I/O调用中使用。内部流对应的内部流需要3秒的时间才会发出onComplete。在第一个流发出onComplete之前，concatMap不订阅与其余元素对应的内部流。一旦完成，下一个流(Observable.just(2).delay(3, TimeUnit.SECONDS))就是subscribed to，依此类推。因此，您可以看到顺序是如何维护的。

关于这两个运算符，你需要记住的是:当元素到达时，flatMap急切地subscribes到内部流。另一方面，在subscribes到下一个流之前，concatMap会等待一个流完成。这就是为什么不能用concatMap进行并行调用的原因。

Answer 2

不怎么有意思。trampoline本质上是在一个线程上执行工作，这些线程以先进先出的顺序调用它的Worker.schedule方法。

在Observable.subscribeOn(Schedulers.trampoline())的情况下，它将是线程订阅，因此应用它没有实际效果。

在Observable.observeOn(Schedulers.trampoline())的情况下，它将是线程信令项，因此也没有实际效果。

concatMap在发送上游项信号的线程或内部Observable完成的线程上执行映射器函数。操作员本质上已经有一个内置的弹床，所以上游的项目和下游的完成不会重叠。在3.x中，将有一个overload接受Scheduler，Schedulers.trampoline()对此也没有实际效果。

Schedulers.trampoline()最好的用例是在不需要异步的单元测试中。因此，您可以参数化您的subscribeOn/observeOn使用，或者使用调度程序挂钩并替换标准的调度程序：

 RxJavaPlugins.setComputationSchedulerHandler(s -> Schedulers.trampoline());
 RxJavaPlugins.setIoSchedulerHandler(s -> Schedulers.trampoline());
 RxJavaPlugins.setNewThreadSchedulerHandler(s -> Schedulers.trampoline());

一旦你完成了，

 RxJavaPlugins.reset();