RxJava2 - 间隔和调度器答案

【问题标题】：RxJava2 - interval and schedulersRxJava2 - 间隔和调度器
【发布时间】：2020-08-24 12:01:38
【问题描述】：

假设我有一个区间，并且我给了它一个计算调度程序。像这样：

Observable
    .interval(0, 1, TimeUnit.SECONDS, computationScheduler)
    .flatMap { ... }

那么，flatmap {...} 中发生的一切是否也会安排在计算线程上？

在 Observable.interval(long initialDelay, long period, TimeUnit unit, Scheduler scheduler) 的源代码中，它说：

 * @param scheduler
 * the Scheduler on which the waiting happens and items are emitted

作为 RxJava 的初学者，我很难理解这条评论。我知道间隔计时器/等待逻辑发生在计算线程上。但是，关于正在发射的项目的最后一部分是否也意味着发射的项目将在同一个线程上使用？还是需要一个observeOn？像这样：

Observable
    .interval(0, 1, TimeUnit.SECONDS, computationScheduler)
    .observeOn(computationScheduler)
    .flatMap { ... }

如果我希望在计算线程上处理发射，是否有必要使用 observeOn？

【问题讨论】：

标签： rx-java rx-java2

【解决方案1】：

这很容易验证：只需打印当前线程即可查看操作符在哪个线程上执行：

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
    .flatMap(e -> {
        System.out.println("on flatmap: " + Thread.currentThread().getName());
        return Observable.just(e).map(x -> "--> " + x);
    })
    .subscribe(s -> {
        System.out.println("on subscribe: " + Thread.currentThread().getName());
        System.out.println(s);
    });

这将始终打印：

on subscribe: main
--> 1
on flatmap: main
on subscribe: main
--> 2
on flatmap: main
on subscribe: main
--> 3
on flatmap: main
on subscribe: main
--> 4
on flatmap: main
on subscribe: main
--> 5
on flatmap: main
on subscribe: main
--> 6
on flatmap: main
on subscribe: main
--> 7
on flatmap: main
on subscribe: main
--> 8
on flatmap: main
on subscribe: main
--> 9

按顺序处理，因为所有都发生在一个线程中 -> main。

observeOn会改变下游执行线程：

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
    .observeOn(Schedulers.computation())
    .flatMap(e -> {
         System.out.println("on flatmap: " + Thread.currentThread().getName());
         return Observable.just(e).map(x -> "--> " + x);
     })
     .observeOn(Schedulers.io())
     .subscribe(s -> {
         System.out.println("on subscribe: " + Thread.currentThread().getName());
         System.out.println(s);
      });

每次执行的结果都不同，但flatmap和subscribe会在不同的线程中处理：

on flatmap: RxComputationThreadPool-1
on subscribe: RxCachedThreadScheduler-1

interval 将充当observeOn 并更改下游执行线程（调度程序）：

Observable.interval(0, 1, TimeUnit.SECONDS, Schedulers.computation())
    .flatMap(e -> {
        System.out.println("on flatmap: " + Thread.currentThread().getName());
        return Observable.just(e).map(x -> "--> " + x);
    })
    .subscribe(s -> {
        System.out.println("on subscribe: " + Thread.currentThread().getName());
        System.out.println(s);
    });

这次执行是在计算调度器的一个线程内顺序执行的：

on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 0
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 1
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 2
on flatmap: RxComputationThreadPool-1
on subscribe: RxComputationThreadPool-1
--> 3
...

interval 默认使用计算调度器，您不需要将其作为参数传递，也不需要observeOn

【讨论】：