RxJava学习指南(三)——调度器

RxJava学习指南

版本：2
整理：陈小默

四、调度器

如果你想给Observable操作符链添加多线程功能，你可以指定操作符（或者特定的Observable）在特定的调度器(Scheduler)上执行。

某些ReactiveX的Observable操作符有一些变体，它们可以接受一个Scheduler参数。这个参数指定操作符将它们的部分或全部任务放在一个特定的调度器上执行。

使用ObserveOn和SubscribeOn操作符，你可以让Observable在一个特定的调度器上执行，ObserveOn指示一个Observable在一个特定的调度器上调用观察者的onNext, onError和onCompleted方法，SubscribeOn更进一步，它指示Observable将全部的处理过程（包括发射数据和通知）放在特定的调度器上执行。

4.1 RxJava示例

4.1.1 调度器的种类

下表展示了RxJava中可用的调度器种类：

4.1.2 默认调度器

在RxJava中，某些Observable操作符的变体允许你设置用于操作执行的调度器，其它的则不在任何特定的调度器上执行，或者在一个指定的默认调度器上执行。下面的表格个列出了一些操作符的默认调度器：

4.1.3 使用调度器

除了将这些调度器传递给RxJava的Observable操作符，你也可以用它们调度你自己的任务。下面的示例展示了Scheduler.Worker的用法：

    with(Schedulers.newThread().createWorker()){
        schedule {
            // your work
        }
        unsubscribe()
    }

4.1.3.1 递归调度器

要调度递归的方法调用，你可以使用schedule，然后再用schedule(this)，示例：

    with(Schedulers.newThread().createWorker()) {
        schedule(object :Action0{
            override fun call() {
                //your work
                schedule(this)
            }
        })
        unsubscribe()
    }

4.1.3.2 检查或设置取消订阅状态

Worker类的对象实现了Subscription接口，使用它的isUnsubscribed和unsubscribe方法，所以你可以在订阅取消时停止任务，或者从正在调度的任务内部取消订阅，示例：

    var QUIT: Boolean = false
    with(Schedulers.newThread().createWorker()) {
        schedule {
            while (!isUnsubscribed) {
                //your work
                if (QUIT) unsubscribe()
            }
        }
    }

Worker同时是Subscription，因此你可以（通常也应该）调用它的unsubscribe方法通知可以挂起任务和释放资源了。

4.1.3.3 延时和周期调度器

你可以使用schedule(action,delayTime,timeUnit)在指定的调度器上延时执行你的任务，下面例子中的任务将在500毫秒之后开始执行：

someScheduler.schedule(someAction, 500, TimeUnit.MILLISECONDS);

使用另一个版本的schedule，schedulePeriodically(action,initialDelay,period,timeUnit)方法让你可以安排一个定期执行的任务，下面例子的任务将在500毫秒之后执行，然后每250毫秒执行一次：

someScheduler.schedulePeriodically(someAction, 500, 250, TimeUnit.MILLISECONDS);

4.1.4 测试调度器

TestScheduler让你可以对调度器的时钟表现进行手动微调。这对依赖精确时间安排的任务的测试很有用处。这个调度器有三个额外的方法：

• advanceTimeTo(time,unit) 向前波动调度器的时钟到一个指定的时间点
• advanceTimeBy(time,unit) 将调度器的时钟向前拨动一个指定的时间段
• triggerActions( ) 开始执行任何计划中的但是未启动的任务，如果它们的计划时间等于或者早于调度器时钟的当前时间

[1]ReactiveX文档