文章目录
- 前言
- RxJava2 线程调度
- RxJava2 怎么进行线程调度
- 总结
前言
经过RxJava2源码分析一，我们了解了RxJava2是怎么进行数据的发送和接收。那么这次我们主要是来分析，RxJava2是如何进行线程调度的。
RxJava2 线程调度
我们知道线程的调度还是很关键的，因为如果正常情况下，android给我们提供的方式是通过Hadler机制来进行线程间通信的，当了解了RxJava2，我们就多了一种通信方式，而且它进行线程间的切换是更加简洁的。
直接通过链式调用，就解决了线程间的切换
RxJava2 怎么进行线程调度
1、线程调度 subscribeOn
subscribeOn这里会指定Observable的调度器，也就是指定事件产生的线程。比如我们的事件是emitter.onNext("1"); 那么指定到哪个线程呢，这里我们指定到Schedulers.io
Schedulers.io( )默认是一个CachedThreadScheduler，很像一个有线程缓存的新线程调度器，这里把它理解成一个IO线程就可以了
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("1");//发送消息，事件产生的地方
emitter.onComplete();
Log.e(TAG, "subscribe called with: e = [" + emitter + "]" + Thread.currentThread());
}
}).subscribeOn(Schedulers.io())
.map(new Function<String, String>() {//这里加了一个map操作符，主要是看事件处理所在的线程
@Override
public String apply(String s) throws Exception {
Log.e(TAG, "apply() called with: s = [" + s + "]" + Thread.currentThread());
return s;
}
})
.subscribeOn(Schedulers.computation())
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG,"onSubscribe() d=" + d );
}
@Override
public void onNext(String s) {
Log.e(TAG,"onNext() s=" + s );
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError() e=" + e );
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete() " );
Log.e(TAG, "onComplete() called with:线程:" + Thread.currentThread());
}
});
我进到subscribeOn 去查看里面的实现
Observable#subscribeOn
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));//主要部分
}
继续看 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))); //主要部分
}
ObservableSubscribeOn#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);//被观察者subscribe 观察者，这部分的任务已经被切换到子线程中了
}
}
SunscribeTask 实现了Runnable，我们看到source.subscribe(parent); 这一步，是完成订阅，而这部分内容是在子线程中处理的，这个时候parent是SubscribeOnObserver
我们看一下SubscribeOnObserver的实现
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {
...
SubscribeOnObserver(Observer<? super T> downstream) {
this.downstream = downstream;
this.upstream = new AtomicReference<Disposable>();
}
@Override
public void onNext(T t) {
downstream.onNext(t);//new Observer中的onNext被调用了
}
@Override
public void onError(Throwable t) {
downstream.onError(t);//new Observer中的onError被调用了
}
@Override
public void onComplete() {
downstream.onComplete();//new Observer中的onComplete被调用了
}
@Override
public void dispose() {
DisposableHelper.dispose(upstream);
DisposableHelper.dispose(this);
}
...
}
看代码中的注释，我们发现完成了new Observer 那几个方法的调用onNext，onError，onComplete。也就在那几个方法中可以接收消息。不同的是这部分内容是在指定的IO线程中进行的。
//接上前面的内容
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
看完SunscribeTask 继续看scheduleDirect
Scheduler#scheduleDirect
public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}
Scheduler#scheduleDirect
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
final Worker w = createWorker();
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
DisposeTask task = new DisposeTask(decoratedRun, w);
w.schedule(task, delay, unit);//开始执行任务
return task;
}
w.schedule(task, delay, unit); 这是最关键部分，在Worker的线程开始执行任务
顺便我们看下DisposeTask，其实它主要是对之前的任务进行包装，以便后面可以方便的取消线程任务，Disposable#dispose（用它取消任务）
static final class DisposeTask implements Disposable, Runnable, SchedulerRunnableIntrospection {
...
DisposeTask(@NonNull Runnable decoratedRun, @NonNull Worker w) {
this.decoratedRun = decoratedRun;
this.w = w;
}
@Override
public void run() {
runner = Thread.currentThread();
try {
decoratedRun.run(); //任务执行
} finally {
dispose();
runner = null;
}
}
...
}
最终我们看下我们的代码执行的结果：
apply() called with: s = [1]Thread[RxCachedThreadScheduler-1,5,main]
onComplete() called with:线程:Thread[RxCachedThreadScheduler-1,5,main]
subscribe() called with: e = [CreateEmitter{DISPOSED}]Thread[RxCachedThreadScheduler-1,5,main]
你可能会问 onComplete()怎么也在IO线程呢，因为我们没有指定Observer在哪里执行，这是接下来要讲的。
2、线程调度 observeOn
observeOn指定observer将会在哪个Scheduler观察这个Observable.
具体点就是：ObserveOn 指定观察者的onNext, onError和onCompleted方法在哪个线程被执行
这次我们增加observeOn(AndroidSchedulers.mainThread())
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("1");
emitter.onComplete();
Log.e(TAG, "subscribe() called with: e = [" + emitter + "]" + Thread.currentThread());
}
}).subscribeOn(Schedulers.io())
.map(new Function<String, String>() {
@Override
public String apply(String s) throws Exception {
//依然是io线程
Log.e(TAG, "apply() called with: s = [" + s + "]" + Thread.currentThread());
return s;
}
})
.observeOn(AndroidSchedulers.mainThread())//新增加部分
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG,"onSubscribe() d=" + d );
}
@Override
public void onNext(String s) {
Log.e(TAG,"onNext() s=" + s );
}
@Override
public void onError(Throwable e) {
Log.e(TAG,"onError() e=" + e );
}
@Override
public void onComplete() {
Log.e(TAG,"onComplete() " );
Log.e(TAG, "onComplete() called with:线程:" + Thread.currentThread());
}
});
我们去看observeOn的实现
Observable#observeOn
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
继续看observeOn
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
ObservableObserveOn #ObservableObserveOn
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;//它就是AndroidSchedulers.mainThread()
this.delayError = delayError;
this.bufferSize = bufferSize;
}
ObservableObserveOn #subscribeActual
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();//scheduler就是上面的AndroidSchedulers.mainThread()
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize)); //主要部分
}
}
这里我们主要看ObserveOnObserver的实现
ObserveOnObserver#schedule
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);//this就是当前的ObserveOnObserver，它实现了Runnable，这把它放到了AndroidSchedulers.mainThread()创建的work 中处理
}
}
}
通过上面我们知道，worker.schedule(this) 把当前任务切换到了主线程中
apply() called with: s = [1]Thread[RxCachedThreadScheduler-1,5,main]
subscribe() called with: e = [CreateEmitter{DISPOSED}]Thread[RxCachedThreadScheduler-1,5,main]
onComplete() called with:线程:Thread[main,5,main]
从日志中，我们能看到onComplete也确实已经被切换到了主线程中执行。
总结
- 我们看到了Observable主要是通过subscribeOn来指定被观察者事件处理所在的线程
- Observer 主要是通过observeOn 来指定观察者的事件处理所在的线程
- 都会在源码中做相应的转换
如果对你有一点点帮助，那是值得高兴的事情。：)
我的csdn：http://blog.csdn.net/shenshizhong
我的简书：http://www.jianshu.com/u/345daf0211ad.、
