減小RxJava中多餘的線程調度

時間 2020-07-20

標籤減小 rxjava 多餘線程調度欄目 Java 简体版

原文原文鏈接

爲何要抑制線程調度

對於一次可觀察序列中的屢次 subscribeOn 或者 observeOn 操做，哪怕指定在相同的 Schedulers.io 調度器上，觀察者操做也會在不一樣的線程上執行，即發生從io線程到io線程的切換。java

這種線程調度是否可避免的呢？react

假如咱們有如下代碼：android

fun fetchItem(): Single<Item> {
    return download(itemId.getAndIncrement())
        .flatMap(::unZip)
        .flatMap(::checkMd5)
}

private fun download(id: Int): Single<Item> {
    return Single.just(id)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
        .map { Item(it) }
}

private fun unZip(item: Item): Single<Item> {
    return Single.just(item)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
}

private fun checkMd5(item: Item): Single<Item> {
    return Single.just(item)
        //Simulate a long time operation
        .delay(300, TimeUnit.MILLISECONDS, Schedulers.io())
}
複製代碼

上面模擬的操做中，download unZip checkMd5 都各自指定了調度器，致使 fetchItem 實際上發生了三次線程切換。對於這種一系列的耗時操做來講，徹底能夠運行在同一條後臺線程上。git

所以就有了 RxSchedulerSuppress 的想法：經過 RxJavaPlugins 替換 Schedulers.IO 調度器的實現—— 若當前操做已經運行在io線程上，那麼就再也不執行io線程池的重複調度。github

實現

經過 RxJavaPlugins 替換 Schedulers.io() 和 Schedulers.compute()ide

RxJava中有五種默認實現的調度器：oop

Schedulers.io()
Schedulers.compute()
Schedulers.single()
Schedulers.trampoline()
Schedulers.newThread()

其中， single 指定在特殊的共享線程上，newThread 指定要建立新線程， trampoline 不會切換線程。因此咱們只可以替換 io 和 compute 這兩個調度器。post

RxJavaPlugins.setIoSchedulerHandler((scheduler)->scheduler);
RxJavaPlugins.setComputationSchedulerHandler((scheduler)->scheduler);

//或者
RxJavaPlugins.setInitIoSchedulerHandler(...);
RxJavaPlugins.setInitComputationSchedulerHandler(...);
//但init這兩個方法必定要在首次使用Schedulers以前調用，
//並且一旦使用就沒法改變，沒法在應用使用過程當中作對比實驗。
複製代碼

實現抑制多餘線程切換的調度器fetch

以IO調度器爲例，咱們徹底無須自行實現和維護線程池，而是應該在原有的調度器實現上作一個切面攔截。經過代理模式，若是當前線程已是io線程，那麼就直接執行，不然就調回原有調度器的實現，進行線程切換。gradle

public final class ImmediateScheduler extends Scheduler {
    //...
    private final Scheduler delegate; //原調度器實現，即Schedulers.io()

    @Override
    @NonNull
    public Disposable scheduleDirect(@NonNull Runnable run) {
        if (predicate(runInCurrentThread)) { 
            //若是已經在io線程，則直接在當前線程運行
            return TrampolineScheduler.instance().scheduleDirect(run);
        } else {
            //不然就用原來的調度器調度
            return delegate.scheduleDirect(run);
        }
    }

    @Override
    @NonNull
    public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
        if (predicate(runInCurrentThread)) {
            //若是已經在io線程，則直接在當前線程運行
            return TrampolineScheduler.instance().scheduleDirect(run, delay, unit);
        } else {
            //不然就用原來的調度器調度
            return delegate.scheduleDirect(run, delay, unit);
        }
    }
    
    //...
}
複製代碼

如何判斷當前線程是否已是io/compute線程

由於io/compute線程實際上是人爲定義的，取決於實際使用的場景，本質上只是線程池配置的不一樣，線程自己沒有這樣的特性。因此我暫時只能依賴 Thread.name 來判斷當前線程所屬的調度器：io調度器的線程以 "RxCachedThreadScheduler" 前綴命名而 compute調度器以 "RxComputationThreadPool" 前綴命名。

若是項目中已經替換過RxJava默認的線程池，就須要根據項目正在使用的ThreadFactory爲線程規範命名，並修改這個當前線程的判斷條件！

效果

抑制 `IO` 到 `IO` 的線程切換

Observable
    .create(emitter -> {
        System.out.println("create on " + Thread.currentThread().getName());
        emitter.onNext("Test");
        emitter.onComplete();
    })
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.io())
    .map(s -> {
        System.out.println("map on " + Thread.currentThread().getName());
        return s;
    })
    .observeOn(Schedulers.io())
    .flatMapCompletable(s -> {
        System.out.println("flatMap on " + Thread.currentThread().getName());
        return Completable.complete();
    })
    .subscribe();
複製代碼

Before Suppress	After Suppress
create on RxCachedThreadScheduler-1 map on RxCachedThreadScheduler-2 flatMap on RxCachedThreadScheduler-3	create on RxCachedThreadScheduler-1 map on RxCachedThreadScheduler-1 flatMap on RxCachedThreadScheduler-1

抑制 `Compute` 到 `IO` 的線程切換

Observable.timer(1, TimeUnit.MILLISECONDS)
    .map(s -> {
        System.out.println("timer on " + Thread.currentThread().getName());
        return s;
    })
    .observeOn(Schedulers.io())
    .subscribe(s ->
        System.out.println("subscribe on " + Thread.currentThread().getName())
    );
複製代碼

Before Suppress	After Suppress
timer on RxComputationThreadPool-1 subscribe on RxCachedThreadScheduler-1	timer on RxComputationThreadPool-1 subscribe on RxComputationThreadPool-1

抑制 `AndroidSchedulers.mainThread` 線程切換

AndroidSchedulers.mainThread().scheduleDirect(runnable);
複製代碼

Before Suppress

//等同於如下代碼
new Handler(Looper.getMainLooper()).post(runnable);
複製代碼

After Suppress

//等同於如下代碼
if (Looper.myLooper() == Looper.getMainLooper()) {
    runnable.run();
} else {
    new Handler(Looper.getMainLooper()).post(runnable);
}
複製代碼

使用

build.gradle添加依賴

dependencies {
    //targeting io.reactivex.schedulers.Schedulers
    implementation 'com.github.YvesCheung.RxSchedulerSuppress:scheduler-suppress:1.0.0'

    //targeting io.reactivex.android.schedulers.AndroidSchedulers
    implementation 'com.github.YvesCheung.RxSchedulerSuppress:scheduler-suppress-android:1.0.0'
}
複製代碼

在使用RxJava以前初始化

public class App extends Application {

    public void onCreate() {
        //...
        
        //抑制經過Schedulers.io()從io線程切換到io線程
        //若是當前線程已是io線程，會當即在當前線程執行
        SchedulerSuppress.SuppressIo();
        //抑制經過Schedulers.compute()從compute線程切換到compute線程
        //若是當前線程已是io線程，會當即在當前線程執行
        SchedulerSuppress.SuppressCompute();
        
        //or
        //抑制經過Schedulers.io()或者Schedulers.compute()
        //從io線程切換到compute線程，或者從compute線程切換到io線程
        SchedulerSuppress.SuppressBackground();

        //抑制經過AndroidSchedulers.mainThread()從main線程拋到下一次Looper循環
        //若是當前線程已是main線程，會當即在當前線程執行
        AndroidSchedulerSuppress.SuppressMain();
        
        //以上代碼須要在 `RxJavaPlugins.lockDown` 前執行
        RxJavaPlugins.lockDown();
    }
}

複製代碼