专栏首页三流程序员的挣扎RxJava 线程调度和源码阅读

RxJava 线程调度和源码阅读

Scheduler

通过 Scheduler 来控制被观察者在哪个线程发射,观察者在哪个线程接收。默认情况,发射时在哪个线程,接收就在哪个线程。

RxJava 内置了几个 Scheduler,通过 Schedulers 来获取。

  • Schedulers.trampoline():当其它排队的任务完成后,在当前线程排队开始执行,FIFO。
  • Schedulers.newThread(): 总是启用新线程,并在新线程执行操作。
  • Schedulers.single():拥有一个线程,和 newThread 相比,这个线程可以共用。
  • Schedulers.computation():计算所使用的 Scheduler。这个计算指的是 CPU 密集型计算,即不会被 I/O 等操作限制性能的操作,例如图形的计算。这个 Scheduler 使用的固定的线程池,大小为 CPU 核数。不要把 I/O 操作放在 computation() 中,否则 I/O 操作的等待时间会浪费 CPU。
  • Schedulers.io():I/O 操作(读写文件、读写数据库、网络信息交互等)所使用的 Scheduler。行为模式和 newThread() 差不多,区别在于 io() 的内部实现是用一个无数量上限的线程池,可以重用空闲的线程,因此多数情况下 io() 比 newThread() 更有效率。不要把计算工作放在 io() 中,可以避免创建不必要的线程。

切换线程

  • subscribeOn(): 控制事件产生的线程。
  • observeOn(): 控制事件消费的线程。
  • unsubscribeOn(): 控制解除订阅时的线程。
Observable.create(ObservableOnSubscribe<String> { emitter ->
                Log.e("RX", "发射线程 ${Thread.currentThread().name}")
                emitter.onNext("a")
            }).subscribeOn(Schedulers.io()) // 在 io 线程发射
                    .observeOn(Schedulers.computation()) // 在计算线程接收
                    .subscribe( {
                        Log.e("RX", "接收线程 ${Thread.currentThread().name}")
                        Log.e("RX", "接收数据 $it")
                    })

日志显示:

发射线程 RxCachedThreadScheduler-1
接收线程 RxComputationThreadPool-1
接收数据 a

如果 observeOn(Schedulers.trampoline()),意思是在当前线程,由于发射时将线程切换到 io 上去了,所以接收时也在这个 io 线程上,日志显示:

发射线程 RxCachedThreadScheduler-1
接收线程 RxCachedThreadScheduler-1
接收数据 a

多次切换接收线程

Observable.create(ObservableOnSubscribe<String> { emitter ->
                Log.e("RX", "发射线程 ${Thread.currentThread().name}")
                emitter.onNext("a")
            }).subscribeOn(Schedulers.io()) // 在 io 线程发射
                    .observeOn(Schedulers.computation())
                    .map {
                        Log.e("RX", "第一次转换数据的线程 ${Thread.currentThread().name}")
                        "$it$it"
                    } // 双倍
                    .observeOn(Schedulers.newThread())
                    .map {
                        Log.e("RX", "第二次转换数据的线程 ${Thread.currentThread().name}")
                        "$it$it" } // 四倍
                    .observeOn(Schedulers.single())
                    .subscribe( {
                        Log.e("RX", "接收线程 ${Thread.currentThread().name}")
                        Log.e("RX", "接收数据 $it")
                    })

日志:

发射线程 RxCachedThreadScheduler-1
第一次转换数据的线程 RxComputationThreadPool-1
第二次转换数据的线程 RxNewThreadScheduler-1
接收线程 RxSingleScheduler-1
接收数据 aaaa

源码分析

对象创建

  1. Observable 通过 create 方法创建 ObservableCreate,将参数 ObservableOnSubscribe 作为 ObservableCreate 的 source。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
   ...
   return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
  1. 在上面返回的 ObservableCreate 上调用 subscribeOn,创建 ObservableSubscribeOn,ObservableCreate 对象作为 source。
public final Observable<T> subscribeOn(Scheduler scheduler) {
    ...
    return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
  1. 在上面返回的 ObservableSubscribeOn 上调用 observeOn,创建 ObservableObserveOn,ObservableSubscribeOn 自己作为 ObservableObserveOn 的 source。
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ...
    return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}

基本上每调一个方法,就创建了一个 Observable 的实现类,然后将上层调用它的 Observable 作为自己内部的 source。

订阅

向上传递观察者

最后调用 ObservableObserveOn 的 subscribe,内部调用 subscribeActual,参数是用户传递进来的 Observer。

protected void subscribeActual(Observer<? super T> observer) {
    // 如果是 TrampolineScheduler,在当前线程执行,不涉及任何线程的切换,所以直接调 source.subscribe
    if (scheduler instanceof TrampolineScheduler) {
        source.subscribe(observer);
    } else {
        Scheduler.Worker w = scheduler.createWorker();
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}

如果 scheduler 不是 TrampolineScheduler,比如 NewThreadScheduler,先调用 createWorker 方法,创建 NewThreadWorker,实现了 Disposable。

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
    private final ScheduledExecutorService executor;

    public NewThreadWorker(ThreadFactory threadFactory) {
        executor = SchedulerPoolFactory.create(threadFactory);
    }

    ...
}

构造中创建一个线程池,再深入下去看,创建了一个只有一个核心线程的线程池对象。然后根据外界传入的 Observer 和这个线程池封装出另一个 Observer。所以在 ObservableSubscribeOn 的对象 source 上用重新封装好的一个观察者订阅它。这样就会调用 ObservableSubscribeOn 的 subscribeActual 方法。

@Override
public void subscribeActual(final Observer<? super T> s) {
    final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

    s.onSubscribe(parent);

    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
  • 第一步对下级的被观察者抛上来的观察者再封装一次,这个 s 其实就是 ObservableObserveOn 中封装后的 ObserveOnObserver。
  • 第二步调用 onSubscribe,看 ObservableObserveOn 的 onSubscribe 方法 public void onSubscribe(Disposable s) { if (DisposableHelper.validate(this.s, s)) { this.s = s; // 创建一个队列 queue = new SpscLinkedArrayQueue<T>(bufferSize); // actual 是下级的观察者,这里最外层用户传进来的那个 Observer 就收到了 onSubscribe 回调 actual.onSubscribe(this); } }
  • 第三步是最主要的开始发射数据

向下发射数据

final class SubscribeTask implements Runnable {
    private final SubscribeOnObserver<T> parent;

    SubscribeTask(SubscribeOnObserver<T> parent) {
        this.parent = parent;
    }

    @Override
    public void run() {
        source.subscribe(parent);
    }
}

SubscribeTask 是一个 Runnable,内部持有最后封装的 Observer。且 run 方法就是用这个观察者去订阅 source。在这个例子里,ObservableSubscribeOn 中的 source 就是上层 ObservableCreate。

而 scheduler.scheduleDirect 这个 Runnable,假设发射的 schedule 是 IoScheduler,scheduleDirect 最后是

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    // 创建线程,返回 EventLoopWorker
    final Worker w = createWorker();

    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    DisposeTask task = new DisposeTask(decoratedRun, w);

    w.schedule(task, delay, unit);

    return task;
}

看最后调用 EventLoopWorker 的 schedule 方法

public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
    ...
    return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
    Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

    ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

    if (parent != null) {
        if (!parent.add(sr)) {
            return sr;
        }
    }

    Future<?> f;
    try {
        if (delayTime <= 0) {
            f = executor.submit((Callable<Object>)sr);
        } else {
            f = executor.schedule((Callable<Object>)sr, delayTime, unit);
        }
        sr.setFuture(f);
    } catch (RejectedExecutionException ex) {
        if (parent != null) {
            parent.remove(sr);
        }
        RxJavaPlugins.onError(ex);
    }

    return sr;
}

这里使用线程池方法执行参数 run 里面的任务,参数这个 run 是 DisposeTask 对象。

public void run() {
    runner = Thread.currentThread();
    try {
        decoratedRun.run();
    } finally {
        dispose();
        runner = null;
    }
}

这个 decoratedRun 就相当于上面的 SubscribeTask。run 方法中是用经过若干层封装的观察者订阅最初的那个被观察者。

观察者收到数据

SubscribeOnObserver 收到 onNext

public void onNext(T t) {
   actual.onNext(t);
}

这样就进了 ObserveOnObserver 的 onNext

@Override
public void onNext(T t) {
    if (done) {
        return;
    }

    if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t);
    }
    schedule();
}

把数据先加到上面创建的那个队列里。

void schedule() {
    // 如果之前是 0
    if (getAndIncrement() == 0) {
        // 这里就是切换接收数据的线程
        worker.schedule(this);
    }
}

线程执行自己的 run 方法

@Override
public void run() {
    if (outputFused) {
        drainFused();
    } else {
        drainNormal();
    }
}
void drainNormal() {
   ...

   final SimpleQueue<T> q = queue;
   final Observer<? super T> a = actual;

   for (;;) {
       ...

       for (;;) {
           ...

           try {
               v = q.poll(); // 取出队列里的值
           } catch (Throwable ex) {
               ...
           }
           ...

           a.onNext(v); // 发射给观察者
       }

       ...
   }
}

// 这个暂时不清楚是何时调用
void drainFused() {
   ...

   for (;;) {
       ...

       actual.onNext(null); // 发射空值

       if (d) {
           ex = error;
           if (ex != null) {
               actual.onError(ex); // 发射事件
           } else {
               actual.onComplete();
           }
           worker.dispose();
           return;
       }

       ...
   }
}

最后将数据发射给最外层的观察者,即这个 actual,同时运行在 observerOn 指定的线程上。

mermaid Observable

mermaid Observer

plantuml

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • RxJava 算术和聚合操作符

    将依次连续发送 observableStr,observableInt,observableBoolean 里的 12 个数据。

    七适散人
  • Dart 笔记 13 - 库

    只有一个表达式的函数,可以简写,类似 Kotlin。=> 后跟着一条表达式,不能是语句,比如 if 语句。

    七适散人
  • Dart 笔记 6 - 函数

    只有一个表达式的函数,可以简写,类似 Kotlin。=> 后跟着一条表达式,不能是语句,比如 if 语句。

    七适散人
  • 每天一道面试题 | day11

    sleep():使一个正在运行的线程处于睡眠状态,是一个静态方法,调用此方法要捕捉InterruptedException异常。

    剑走天涯
  • 线程的实现方式

    本文主要介绍线程的3种实现方式和java线程的实现方式。线程是比进程更轻量级的调度执行单位,线程的引入,可以把一个进程的资源分配 和执行调度分开,各个进程既可以...

    栋先生
  • Java并发编程实战 05等待-通知机制和活跃性问题

    Java并发编程实战 01并发编程的Bug源头 Java并发编程实战 02Java如何解决可见性和有序性问题 Java并发编程实战 03互斥锁 解决原子性问...

    Johnson木木
  • 一篇文章搞懂Python多线程简单实现和GIL

    今天开始打算开一个新系列,就是python的多线程和多进程实现,这部分可能有些新手还是比较模糊的,都知道python中的多线程是假的,但是又不知道怎么回事,首先...

    南山烟雨
  • CLR线程池边缘情况测试

    设置工作者线程的下限数量5,上限数量10,给线程池一瞬间安排20个任务,每个任务占用线程时间18秒,然后观察线程池在对待线程数量的上下限的反应,以及完成一个任务...

    小蜜蜂
  • Linux下的进程类别(内核线程、轻量级进程和用户进程)--Linux进程的管理与调度(四)

    虽然我们在区分Linux进程类别, 但是我还是想说Linux下只有一种类型的进程,那就是task_struct,当然我也想说linux其实也没有线程的概念, 只...

    233333
  • 死磕 java同步系列之开篇

    同步系列,这是彤哥想了好久的名字,本来是准备写锁相关的内容,但是java中的CountDownLatch、Semaphore、CyclicBarrier这些类又...

    彤哥

扫码关注云+社区

领取腾讯云代金券