阅读 194

Android的第三方库-RxJava:RxJava的线程切换

介绍

RxJava的基础使用和基础流程可以看看上一篇的文章《Android的三方库 - RxJava:RxJava的使用和基本订阅流程》

实际项目中经常会有一些数据获取操作,这就需要使用到RxJava的线程了。 所以让我们来看看RxJava的线程切换。

1.线程切换案例

首先看一个小的案例:

Log.i(TAG, "当前线程 : " + Thread.currentThread().getName());
Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        Log.i(TAG, "subscribe: currentThread : " + Thread.currentThread().getName());
        emitter.onNext("ONE");
        emitter.onComplete();
    }
}).subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) {
        Log.i(TAG, "onSubscribe: currentThread : " + Thread.currentThread().getName());
    }

    @Override
    public void onNext(String s) {
        Log.i(TAG, "onNext: currentThread : " + Thread.currentThread().getName());
        Log.i(TAG, "onNext: s : " +  s);
    }

    @Override
    public void onError(Throwable e) {
    
    }

    @Override
    public void onComplete() {
        Log.i(TAG, "onComplete: currentThread : " + Thread.currentThread().getName());
    }
});
复制代码

输出结果:

当前线程 : main
onSubscribe: currentThread : main
subscribe: currentThread : RxCachedThreadScheduler-1
onNext: currentThread : main
onNext: s : ONE
onComplete: currentThread : main
复制代码

然后我们在线程中来执行这段代码:

new Thread(new Runnable() {
    @Override
    public void run() {
        RxObservable3(); // 这个就是上面那段代码
    }
}).start();
复制代码

输出结果:

当前线程 : Thread-5
onSubscribe: currentThread : Thread-5
subscribe: currentThread : RxCachedThreadScheduler-1
onNext: currentThread : main
onNext: s : ONE
onComplete: currentThread : main
复制代码

我们可以很明显得出几个结论:

1. onSubscribe运行的线程和代码运行所在的线程是一致的。
2. Observable 运行的线程是subscribeOn指定的线程。
3. Observer 运行的线程是 observeOn 指定的线程。
复制代码

2.源码分析

现在我们对源码来进行分析。

2.1 subscribeOn()

在我们的案例中的调用:

subscribeOn(Schedulers.io())
复制代码

进入到它的源码中:

public final Observable<T> subscribeOn(Scheduler scheduler) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
复制代码

在这里传入的参数 this 其实就是 当前创建的Observable( 就是 ObservableCreate)将它封装成 ObservableSubscribeOn,并返回它的对象。

scheduler就是你使用的 Schedulers.io()

我们进入到 io()

public static Scheduler io() {
    return RxJavaPlugins.onIoScheduler(IO);
}
复制代码

在这里的 IO 是一个对象,它的具体实现要看 IOTask

static final class IOTask implements Callable<Scheduler> {
    @Override
    public Scheduler call() throws Exception {
        return IoHolder.DEFAULT;
    }
}
--------
static final class IoHolder {
    static final Scheduler DEFAULT = new IoScheduler();
}

复制代码

可以看到我们就这样得到了IoScheduler的对象。先静置这里,等会使用。

2.1.1 ObservableSubscribeOn#subscribeActual( )

我们根据上一篇的文章的分析,知道subscribeActual()是一个抽象方法,之前的实现是ObservableCreate中,

现在ObservableCreate对象被包装成为一个新的ObservableSubscribeOn对象

因此我们来看看 ObservableSubscribeOn 中的 subscribeActual方法:

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

    observer.onSubscribe(parent);
    
    parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}
复制代码

目前看到都是没有使用过任何其他的线程,所以 观察者( observer ) 的 onSubscribe() 运行的线程就是当前的线程。

所以 onSubscribe()执行的线程就是 当前的线程。

即:

onSubscribe: currentThread : Thread-5
复制代码

继续来看下一行代码,执行了 SubscribeTask 这个类。

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

    SubscribeTask(SubscribeOnObserver<T> parent) {
        this.parent = parent;
    }
    
    @Override
    public void run() {
        source.subscribe(parent);
    }
}

复制代码

这是一个Runnable类,在run方法中 执行 subscribe 方法,

这里的source其实就是上一个Observable,也就是ObservableCreate的对象。

2.1.2 Schedule#scheduleDirect()

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
    return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}

-----------------------------------

@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
    // 这个createWorker 是一个抽象方法,具体的实现是在 Schedule的子类中,在这里也就是IoScheduler
    final Worker w = createWorker();
    //  decoratedRun实际上还是SubscribeTask这个对象
    final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
    // 将 Workder 和 Runnable 包装成为一个DisposeTask
    DisposeTask task = new DisposeTask(decoratedRun, w);
    // 执行
    w.schedule(task, delay, unit);

    return task;
}
复制代码

再继续看一下IoScheduler中的createWorker()workerschedule()方法。

final AtomicReference<CachedWorkerPool> pool; // AtomicReference 就是实现对象引用的原子更新

@NonNull
@Override
public Worker createWorker() {
    // 得到一个EventLoopWorker对象,并传入一个缓存池
    return new EventLoopWorker(pool.get());
}

--------------------------

static final class EventLoopWorker extends Scheduler.Worker {
    private final CompositeDisposable tasks;
    private final CachedWorkerPool pool;
    private final ThreadWorker threadWorker;

    final AtomicBoolean once = new AtomicBoolean();

    EventLoopWorker(CachedWorkerPool pool) {
        this.pool = pool;
        this.tasks = new CompositeDisposable();
        this.threadWorker = pool.get();
    }

    .... //省略无关代码

    @NonNull
    @Override
    public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
        if (tasks.isDisposed()) {
            // don't schedule, we are unsubscribed
            return EmptyDisposable.INSTANCE;
        }
        // Runnable 最终是交给了threadWorker 去执行
        return threadWorker.scheduleActual(action, delayTime, unit, tasks);
    }
}
复制代码

我们可以看下这个 ThreadWorker ,它是没有实现 scheduleActual()这个方法的,

我们来到它的父类 NewThreadWorker

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

@NonNull
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
     // 这个run 是SubscribeTask
    Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
    // 继续封装成为ScheduledRunnable
    ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
    .... // 省略无关代码
    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;
}
复制代码

最后就是 executor这个线程池对象来执行任务,SubscribeTask会被线程池执行,

也就是说 Observablesubscribe()方法会在IO线程中被调用。

之前的输出结果:

subscribe: currentThread : RxCachedThreadScheduler-1
复制代码

所以 subscribe() 执行的线程就是 subscribeOn 指定的线程(在这里就是IoScheduler)。

2.1.3 多次设置subscrineOn()的问题

案例:

        //省略前后代码,看重点部分
        .subscribeOn(Schedulers.io())//第一次
        .subscribeOn(Schedulers.newThread())//第二次
        .subscribeOn(AndroidSchedulers.mainThread())//第三次
复制代码

输出结果:

subscribe: currentThread : RxCachedThreadScheduler-1 
复制代码

就是第一次的 subscribeOn()的设置起作用了,这是为什么呢?

因为 每一次调用 subscribeOn() 都会将之前的Observable重新包装。

我们来看这张图: 引用 玉刚说的 详解 RxJava 的消息订阅和线程切换原理中的图

从第三次的ObservableSubscribeOn每次都会通知它的上一个Obsevable

最后都会上传到第一次的ObservableSubscribeOn中,

所以不过设置多少,都只有第一次的subscribeOn() 才是生效的。

2.2 observeOn( )

在案例中的设置:

.observeOn(AndroidSchedulers.mainThread())
复制代码

进入源码:

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> observeOn(Scheduler scheduler) {
    return observeOn(scheduler, false, bufferSize());
}

-----------------

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    ObjectHelper.requireNonNull(scheduler, "scheduler is null");
    ObjectHelper.verifyPositive(bufferSize, "bufferSize");
    return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
复制代码

这个也是重新包装了一个ObservableObserveOn对象,

这里的this就是之前被封装过的 ObservableSubscribeOn 对象。

然后我们继续进入到ObservableObserveOn 中去。

2.2.1 ObservableObserveOn#subscribeActual( )

@Override
protected void subscribeActual(Observer<? super T> observer) {
    // 如果是当前线程的话,直接调用ObservableSubscribeOn的subscribe()方法
    if (scheduler instanceof TrampolineScheduler) {
        source.subscribe(observer);
    } else {
        // 这里的 scheduler 其实就是 AndroidSchedulers.mainThread()
        Scheduler.Worker w = scheduler.createWorker();
        // 将worker封装成ObserveOnObserver
        // 这里的source.subscribe是在IO线程中执行
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}
复制代码

接下来我们来看看 ObserveOnObserver 中的源码:

2.2.2 ObserveOnObserver#onNext()

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

    if (sourceMode != QueueDisposable.ASYNC) {
        queue.offer(t); // 加入队列中
    }
    schedule();
}
复制代码

主要是执行 schedule()

2.2.3 ObserveOnObserver#schedule()

void schedule() {
    if (getAndIncrement() == 0) {
        worker.schedule(this);
    }
}
复制代码

其实 ObserveOnObserver 自己也实现了 Runnable ,所以就是调用了自己。

这里的worker 就是 你传入的 主线程(mainThread

然后我们来看看 run 方法:

@Override
public void run() {
    if (outputFused) { // outputFused 默认是false
        drainFused();
    } else {
        drainNormal();
    }
}
复制代码

进入到 drainNormal() 中:

void drainNormal() {
    int missed = 1;
    // 这里就是存储消息的队列
    final SimpleQueue<T> q = queue;
    // 这里是  downstream 就是 自定义的Observer
    final Observer<? super T> a = downstream;
   
      .... //省略部分代码 
      // 队列中取出消息
                v = q.poll();
     .... //省略部分代码
      // 这里就是调用自定义的Observer#onNext()
            a.onNext(v);
      ....
    }
}
复制代码

这样我们最终会调用到 我们自定义的Observer#onNext()

所以Observer#onNext() 是在 observeOn() 指定的线程中调用的。

注意:

在这里的包装顺序是

CreateEmitter包装了SubscribeOnObserver

SubscribeOnObserver包装了ObserveOnObserver

ObserveOnObserver包装了自定义的Observer

对于调用onNext() 方法:

我们在subscribe 中使用 ObservableEmitter 中调用了onNext()

会在SubscribeOnObserver中调用onNext()

然后会继续调用到ObserveOnObserver#onNext()

所以在run中调用的Observer的对象其实是自定义的Observer

3.结束

以上就是我对于RxJava线程切换的源码分析,如有不对,请指正。

关注下面的标签,发现更多相似文章
评论