如何使用Schedulers.trampoline()在rxjava

Question

由于Schedulers.trampoline()使作业在当前线程上工作，我找不到Schedulers.trampoline()和没有Schedulers设置的情况之间的区别。

使用Schedulers.trampoline():

Observable.from(1, 2, 3)
    .observeOn(Schedulers.trampoline())
    .subscribe(System.out::println)

不使用调度器:

Observable.from(1, 2, 3)
    .subscribe(System.out::println)

我认为以上代码的作用是相同的。我真的很想知道为什么Schedulers.trampoline()存在于RxJava的API中。

在什么情况下，我应该使用Schedulers.trampoline()

Answer 1

在observeOn/subscribeOn中使用调度器不会获得任何好处。您可以使用调度程序中的Worker来安排下班后的工作。

请看下面的例子。我使用的是RxJava2-RC5

@Test
public void trampoline() throws Exception {
        Scheduler scheduler = Schedulers.trampoline();
        Scheduler.Worker worker = scheduler.createWorker();
        Runnable r1 = () -> {
            System.out.println("Start: r1");
            System.out.println("End: r1");
        };
        Runnable r2 = () -> {
            System.out.println("Start: r2");
            worker.schedule(r1);
            System.out.println("End: r2");
        };
        worker.schedule(r2);
}

输出:

Start: r2 End: r2 Start: r1 End: r1

如果你递归地调度工作，蹦床工人会派上用场，因为你不会得到和StackOverFlow。

这个例子是从introtorx (http://www.introtorx.com/content/v1.0.10621.0/15_SchedulingAndThreading.html)转换成RxJava的

Answer 2

Schedulers.trampoline()还有另一种用法，请查看以下内容:

println("Current thread: ${Thread.currentThread()}")
Observable.interval(500, TimeUnit.MILLISECONDS, Schedulers.trampoline())
    .subscribe{
        println("$it thread: ${Thread.currentThread()}")
    }
println("This never will be reached")

如果在主线程上运行代码，那么结果将是这样的:

Current thread: Thread[main,5,main]
0 thread: Thread[main,5,main]
1 thread: Thread[main,5,main]
2 thread: Thread[main,5,main]
3 thread: Thread[main,5,main]
4 thread: Thread[main,5,main]
5 thread: Thread[main,5,main]
...

和println("This never will be reached")将永远不会到达。

Answer 3

姗姗来迟，但来自文档:

即时调度程序在2.x中不存在。这经常发生误用并且没有正确实现Scheduler规范无论如何;它包含阻止睡眠延迟动作，但没有完全不支持递归调度。使用Schedulers.trampoline ()相反。

这意味着你接受的答案可能是错误的。看到https://medium.com/@I_Love_Coding/rxjava-schedulers-trampoline-use-cases-283f6649cbf说:

所有订阅trampoline()的作业将被排队并执行一个接一个。

显示了一个像

这样的例子

 .subscribeOn(Schedulers.trampoline())

Answer 4

对于测试也很方便。在这种情况下，您不直接使用调度器，而是将一个对象(或提供程序)注入(或通过构造函数设置或传入类)到您的类中，该对象(或提供程序)包含所需的调度器。在subscribeOn和observeOn中使用这个对象保存调度程序。当测试时，你注入(或设置或手动)trampoline()，这样你的测试都在同一个线程上运行。