如果你想給Observable操作符鏈添加多線程功能，你可以指定操作符（或者特定的Observable）在特定的調(diào)度器(Scheduler)上執(zhí)行。

某些ReactiveX的Observable操作符有一些變體，它們可以接受一個(gè)Scheduler參數(shù)。這個(gè)參數(shù)指定操作符將它們的部分或全部任務(wù)放在一個(gè)特定的調(diào)度器上執(zhí)行。

使用ObserveOn和SubscribeOn操作符，你可以讓Observable在一個(gè)特定的調(diào)度器上執(zhí)行，ObserveOn指示一個(gè)Observable在一個(gè)特定的調(diào)度器上調(diào)用觀察者的onNext, onError和onCompleted方法，SubscribeOn更進(jìn)一步，它指示Observable將全部的處理過(guò)程（包括發(fā)射數(shù)據(jù)和通知）放在特定的調(diào)度器上執(zhí)行。

RxJava示例

調(diào)度器的種類

下表展示了RxJava中可用的調(diào)度器種類：

默認(rèn)調(diào)度器

在RxJava中，某些Observable操作符的變體允許你設(shè)置用于操作執(zhí)行的調(diào)度器，其它的則不在任何特定的調(diào)度器上執(zhí)行，或者在一個(gè)指定的默認(rèn)調(diào)度器上執(zhí)行。下面的表格個(gè)列出了一些操作符的默認(rèn)調(diào)度器：

使用調(diào)度器

除了將這些調(diào)度器傳遞給RxJava的Observable操作符，你也可以用它們調(diào)度你自己的任務(wù)。下面的示例展示了Scheduler.Worker的用法：

worker = Schedulers.newThread().createWorker();
worker.schedule(new Action0() {

    @Override
    public void call() {
        yourWork();
    }

});
// some time later...
worker.unsubscribe();

遞歸調(diào)度器

要調(diào)度遞歸的方法調(diào)用，你可以使用schedule，然后再用schedule(this)，示例：

worker = Schedulers.newThread().createWorker();
worker.schedule(new Action0() {

    @Override
    public void call() {
        yourWork();
        // recurse until unsubscribed (schedule will do nothing if unsubscribed)
        worker.schedule(this);
    }

});
// some time later...
worker.unsubscribe();

檢查或設(shè)置取消訂閱狀態(tài)

Worker類的對(duì)象實(shí)現(xiàn)了Subscription接口，使用它的isUnsubscribed和unsubscribe方法，所以你可以在訂閱取消時(shí)停止任務(wù)，或者從正在調(diào)度的任務(wù)內(nèi)部取消訂閱，示例：

Worker worker = Schedulers.newThread().createWorker();
Subscription mySubscription = worker.schedule(new Action0() {

    @Override
    public void call() {
        while(!worker.isUnsubscribed()) {
            status = yourWork();
            if(QUIT == status) { worker.unsubscribe(); }
        }
    }

});

Worker同時(shí)是Subscription，因此你可以（通常也應(yīng)該）調(diào)用它的unsubscribe方法通知可以掛起任務(wù)和釋放資源了。

延時(shí)和周期調(diào)度器

你可以使用schedule(action,delayTime,timeUnit)在指定的調(diào)度器上延時(shí)執(zhí)行你的任務(wù)，下面例子中的任務(wù)將在500毫秒之后開始執(zhí)行：

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

使用另一個(gè)版本的schedule，schedulePeriodically(action,initialDelay,period,timeUnit)方法讓你可以安排一個(gè)定期執(zhí)行的任務(wù)，下面例子的任務(wù)將在500毫秒之后執(zhí)行，然后每250毫秒執(zhí)行一次：

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

測(cè)試調(diào)度器

TestScheduler讓你可以對(duì)調(diào)度器的時(shí)鐘表現(xiàn)進(jìn)行手動(dòng)微調(diào)。這對(duì)依賴精確時(shí)間安排的任務(wù)的測(cè)試很有用處。這個(gè)調(diào)度器有三個(gè)額外的方法：

• advanceTimeTo(time,unit) 向前波動(dòng)調(diào)度器的時(shí)鐘到一個(gè)指定的時(shí)間點(diǎn)
• advanceTimeBy(time,unit) 將調(diào)度器的時(shí)鐘向前撥動(dòng)一個(gè)指定的時(shí)間段
• triggerActions(?) 開始執(zhí)行任何計(jì)劃中的但是未啟動(dòng)的任務(wù)，如果它們的計(jì)劃時(shí)間等于或者早于調(diào)度器時(shí)鐘的當(dāng)前時(shí)間