减小RxJava中多余的线程调度

为何要抑制线程调度

对于一次可观察序列中的屡次 subscribeOn 或者 observeOn 操做，哪怕指定在相同的 Schedulers.io 调度器上，观察者操做也会在不一样的线程上执行，即发生从io线程到io线程的切换。

这种线程调度是否可避免的呢？

假如咱们有如下代码：

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 实际上发生了三次线程切换。 对于这种一系列的耗时操做来讲，彻底能够运行在同一条后台线程上。

所以就有了 RxSchedulerSuppress 的想法：经过 RxJavaPlugins 替换 Schedulers.IO 调度器的实现—— 若当前操做已经运行在io线程上，那么就再也不执行io线程池的重复调度。

实现

1. 经过 RxJavaPlugins 替换 Schedulers.io() 和 Schedulers.compute()

   RxJava中有五种默认实现的调度器：

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

   其中， single 指定在特殊的共享线程上，newThread 指定要建立新线程， trampoline 不会切换线程。因此咱们只可以替换 io 和 compute 这两个调度器。

   RxJavaPlugins.setIoSchedulerHandler((scheduler)->scheduler);
   RxJavaPlugins.setComputationSchedulerHandler((scheduler)->scheduler);
   
   //或者
   RxJavaPlugins.setInitIoSchedulerHandler(...);
   RxJavaPlugins.setInitComputationSchedulerHandler(...);
   //但init这两个方法必定要在首次使用Schedulers以前调用，
   //并且一旦使用就没法改变，没法在应用使用过程当中作对比实验。
   复制代码

2. 实现抑制多余线程切换的调度器

   以IO调度器为例，咱们彻底无须自行实现和维护线程池，而是应该在原有的调度器实现上作一个切面拦截。经过代理模式，若是当前线程已是io线程，那么就直接执行，不然就调回原有调度器的实现，进行线程切换。

   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);
           }
       }
       
       //...
   }
   复制代码

3. 如何判断当前线程是否已是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);
}
复制代码

使用

1. 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'
}
复制代码

2. 在使用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();
    }
}

复制代码

项目源码

欢迎交流或者提交您的修改 github.com/YvesCheung/…