好文推荐&＃xff1a;
作者&＃xff1a;xiangcman
转载&＃xff1a;https://juejin.cn/post/6900870262062120967
大家都知道RxJava上手是非常难的一个框架&＃xff0c;为什么说是难呢&＃xff0c;因为它的功能非常强大&＃xff0c;各种操作符让人很难上手&＃xff0c;搭配使用带生命周期的框架有RxLife等。以至于后面出了很多类似Rxjava的框架&＃xff0c;有RxAndroid&＃xff0c;我们用的RxJava切换主线程就是出自该框架&＃xff0c;后面ACC架构中有LiveData、Lifecycle、包括后面协成中出来的flow异步流&＃xff0c;都是向Rxjava靠拢&＃xff0c;不过ACC出来的这些框架他们最大的特点就是操作简单&＃xff0c;上手简单。所以RxJava强大的框架背后读懂源码是非常难的&＃xff0c;因此秉着扒开源码的想法&＃xff0c;探索一些RxJava平时的疑惑。
在Rxjava里面有几个角色我们需要弄明白&＃xff1a;
(Observer)的订阅&＃xff0c;然后通过发射器(Emitter)发射数据给订阅者。Disposable传给订阅者&＃xff0c;然后在被订阅者中的发射器(Emitter)发射数据给订阅者(Observer)。(Observer)&＃xff0c;然后在发射器相应的方法把数据传给订阅者(Observer)。Observer的一种变体&＃xff0c;Observer的每一个方法都会对应一个Consumer&＃xff0c;比如Observer的onNext、onError、onComplete、onSubscribe都会对应一个Consumer。Disposable&＃xff0c;一个是在Observer的onSubscribe方法回调回来&＃xff0c;第二个是在subscribe订阅方法传consumer的时候会返回。这里以一种最基本的订阅来介绍它们之间的关系&＃xff1a;
Observer到Observable订阅的过程&＃xff0c;首先是下游的Observer发起Observable的subscribe方法&＃xff0c;而该方法会调用到Observable的subscribeActual&＃xff0c;并且会把下游的Observer传给该方法中&＃xff0c;这就是上面的过程①。Emitter对象&＃xff0c;该发射器需要接受过程①中的Observer&＃xff0c;也就是下游的观察者&＃xff0c;接着给下游的Observer的添加订阅的监听&＃xff0c;也就是onSubscribe方法&＃xff0c;并且把Disposable传给onSubscribe方法&＃xff0c;这里的Disposable其实是刚才创建的发射器&＃xff0c;因为本身发射器也是实现了Disposable类型&＃xff0c;所以下游的Observer会回调到上游的Disposable。Observer订阅监听后&＃xff0c;继续向上一层的Observable添加订阅&＃xff0c;也就是把②中创建的发射器或②中创建的Observer传给上游的Observable的subscribe方法。observer到上游的Observable订阅的过程&＃xff0c;接着就是发射数据了&＃xff0c;由于最上游的Observable会在subscribe方法中收到发射器&＃xff0c;因此我们可以利用发射器把数据发送到下游的Observer&＃xff0c;也就是onNext、onError、onComplete等方法。该篇文章会通过源码的形式介绍一下几点&＃xff0c;以及总结面试过程中如何应答RxJava的问题
通常我们写一个从订阅到发送数据的示例如下:
Observable.create(new ObservableOnSubscribe
}).subscribe(new Observer
});
相信这是最简单的事件发送的示例&＃xff0c;这没什么好说的&＃xff0c;那它们是怎么发送数据&＃xff0c;接收数据的呢&＃xff0c;下面我会把代码拆分来看&＃xff0c;因为现在是链式调用&＃xff0c;我把代码拆分如下:
这里把创建observable和observer、以及发起订阅分别拆开来写&＃xff0c;后面方便我们分析代码&＃xff0c;首先是第一步发起订阅observable.subscribe(observer)&＃xff1a;
&＃64;Override
public final void subscribe(&＃64;NonNull Observer observer) {try {subscribeActual(observer);} catch (NullPointerException e) { // NOPMDthrow e;} catch (Throwable e) {NullPointerException npe &＃61; new NullPointerException("Actually not, but can&＃39;t throw other exceptions due to RS");throw npe;}
}
observable的订阅方法关键一句subscribeActual(observer)&＃xff0c;这里提一句&＃xff0c;所有的被观察者的订阅入口都是subscribeActual方法&＃xff0c;而subscribeActual在被观察者中是抽象方法&＃xff0c;因此看对应的observable子类实现的逻辑&＃xff0c;在上面通过Observable.create创建的被观察者是ObservableCreate&＃xff0c;它是Observable的子类&＃xff0c;我么需要明确&＃xff0c;RxJava中的操作符都会对应一个Observable的子类&＃xff0c;比如just操作符对应的是ObservableJust的被观察者&＃xff0c;好了&＃xff0c;我们看ObservableCreate的subscribeActual实现&＃xff1a;
&＃64;Override
protected void subscribeActual(Observer observer) {//创建发射器&＃xff0c;并且把下游的observer给发射器CreateEmitter
}
先是创建CreateEmitter类型的发射器&＃xff0c;把下游的observer传给发射器&＃xff0c;注意此处的发射器是实现了Disposable接口&＃xff0c;所以紧接着会把发射器通过下游的观察者的onSubscribe方法传给下游观察者&＃xff0c;注意此处传的是Disposable对象。接着会给上游的ObservableOnSubscribe添加订阅&＃xff0c;并且把下游的observer给上游的ObservableOnSubscribe。 为了描述订阅的过程&＃xff0c;我们画一张时序图&＃xff1a;
订阅是从下游的Observer向上游的Observable发送订阅&＃xff0c;然后在订阅的过程中&＃xff0c;给下游的Observer发送订阅监听&＃xff0c;并且给上游的被观察者添加订阅。
上面我们知道在ObservableCreate的subscribeActual方法中给上游的ObservableOnSubscribe添加了onSubscribe订阅过程&＃xff0c;并且把当前的发射器传给了ObservableOnSubscribe&＃xff0c;而在我们上面的示例中定义的ObservableOnSubscribe内部类的subscribe方法通过传过来的发射器添加了如下代码:
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
所以到这里可以看到是通过发射器的onNext和onComplete发送数据&＃xff0c;而emitter是上面订阅过程传过来的CreateEmitter&＃xff0c;所以直接看它的onNext和onComplete&＃xff1a;
&＃64;Override
public void onNext(T t) {if (t &＃61;&＃61; null) {onError(ExceptionHelper.createNullPointerException("onNext called with a null value."));return;}//如果isDisposed为false&＃xff0c;则可以继续发送数据if (!isDisposed()) {observer.onNext(t);}
}
很简单&＃xff0c;给observer发送数据&＃xff0c;而当前的observer是订阅过程中传进来的下游observer&＃xff0c;所以大家明白了吧&＃xff0c;最终是下游的observer接收到数据。
发送主要通过上游的被观察者通知发射器&＃xff0c;然后发射器会发送给下游的observer。
上面我们看到emitter.onNext三次完了后&＃xff0c;会发送onComplete事件&＃xff0c;那onComplete处理啥呢&＃xff1a;
&＃64;Override
public void onComplete() {if (!isDisposed()) {try {observer.onComplete();} finally {dispose();}}
}
这是发射器中onComplete的定义&＃xff0c;dispose方法是控制是否还能发送数据&＃xff0c;其实这里的 CreateEmitter它是一个AtomicReference原子类包装Disposable的实现类&＃xff0c;而我们dispose方法正是将该原子类添加了常量的DISPOSED&＃xff0c;而在onNext方法中通过判断isDisposed是否为false才能继续发送数据。而isDisposed什么时候为false呢&＃xff1f;当AtomicReference中的包装对象不是DISPOSED。所以我们的onComplete是用来控制不能发送数据的。
您可以通过如下代码测试&＃xff1a;
emitter.onNext(1);
emitter.onNext(2);
emitter.onComplete();
emitter.onNext(3);
看看下游的observer是否还能收到3的数据。
onComplete是用来控制不能发送数据的&＃xff0c;也就是不能onNext了&＃xff0c;包括onError也是不能再发送onNext数据了&＃xff0c;该方法中也是调用了dispose方法。
map和flatMap是我们经常用的转换操作&＃xff0c;我们先看看map如何使用&＃xff1a;
Observable
}
通过createObservable的map操作生成了一个mapObservable的被观察者&＃xff0c;最终通过mapObservable与observer形成订阅关系&＃xff0c;而map操作需要一个Function的接口&＃xff0c;第一个泛型是入参类型&＃xff0c;第二个泛型是出参的类型&＃xff0c;也就是apply的返回值&＃xff0c;这里定义map的出参类型是String类型。 我们再来看下flatMap如何使用&＃xff1a;
Observable
});
flatMapObservable.subscribe(observer);
在上面的mapObservable基础上通过flatMap返回flatMapObservable&＃xff0c;最后通过flatMapObservable订阅observer。flatMap的Function第二个泛型是ObservableSource类型的&＃xff0c;Observable的父类是ObservableSource类型&＃xff0c;因此第二个参数返回Observable也可以。
从上面可以看出map是通过原始数据类型返回另外一种数据类型&＃xff0c;而flatMap是通过原始数据类型返回另外一种被观察者。
关于面试也有问flatMap和concatMap的区别&＃xff0c;下面我通过一个例子来演示他们的区别:
Observable
Observable
});
Observable
Observer
};
observeOnObservable.subscribe(observer);
在上面flatMap操作过程中为了演示flatMap和concatMap的区别&＃xff0c;在数据为2的时候让返回的observable延迟500毫秒&＃xff0c;我们看到的结果如下:
上面例子中3是由2的发射数据发射过来的&＃xff0c;而正好数据为2的时候让延迟了500毫秒&＃xff0c;那如果换成concatMap结果是按照发射数据的顺序来返回的。
concatMap和flatMap的功能是一样的&＃xff0c; 将一个发射数据的Observable变换为多个Observables&＃xff0c;然后将它们发射的数据放进一个单独的Observable。只不过最后合并ObservablesflatMap采用的merge&＃xff0c;而concatMap采用的是连接(concat)。总之一句一话,他们的区别在于&＃xff1a;concatMap是有序的&＃xff0c;flatMap是无序的&＃xff0c;concatMap最终输出的顺序与原序列保持一致&＃xff0c;而flatMap则不一定&＃xff0c;有可能出现交错。
关于其他的操作符比如merge、concat、zip都是合并&＃xff0c;interval是周期执行&＃xff0c;timer是延迟发送数据。如果要学习更多的操作符请猛戳官网
其实想知道它们的区别&＃xff0c;我们直接看对应的Observer的方法有哪些:
Maybe从字面意思是可能的意思&＃xff0c;看下MaybeObserver接口&＃xff1a;
public interface MaybeObserver<&＃64;NonNull T> { void onSubscribe(&＃64;NonNull Disposable d);void onSuccess(&＃64;NonNull T t); void onError(&＃64;NonNull Throwable e);void onComplete();
}
它没有onNext方法&＃xff0c;也就是说不能发多条数据&＃xff0c;如果回调到onSuccess再不能发消息了&＃xff0c;如果直接回调onComplete相当于没发数据&＃xff0c;也就是说Maybe可能不发送数据&＃xff0c;如果发送数据只会发送单条数据。
这个不用多说了&＃xff0c;它是能发送多条数据的&＃xff0c;直到发送onError或onComplete才不会再发送数据了&＃xff0c;当然它也是可以不发送数据的&＃xff0c;直接发送onError或onComplete。
public interface SingleObserver<&＃64;NonNull T> {void onSubscribe(&＃64;NonNull Disposable d);void onSuccess(&＃64;NonNull T t);void onError(&＃64;NonNull Throwable e);
}
single也是发送单条数据&＃xff0c;单是它要么成功要么失败。
Flowable没有FlowableObserver接口&＃xff0c;它是由FlowableSubscriber代表观察者&＃xff0c;Flowable在后面被压的时候讲&＃xff0c;我们只要知道它是被压策略的一个被观察者。
public interface CompletableObserver {void onSubscribe(&＃64;NonNull Disposable d);void onComplete();void onError(&＃64;NonNull Throwable e);
}
Completable不发送数据&＃xff0c;只会发送成功或失败的事件&＃xff0c;当然这个用得很少。
从上面各个对应的observer接口来看&＃xff0c;如果只想发一条数据&＃xff0c;或者不发数据就用Maybe&＃xff0c;如果想法多条数据或者不发数据就用Observable&＃xff0c;如果只发一条数据或者失败就用Single&＃xff0c;如果想用背压策略使用Flowable&＃xff0c;如果不发数据就用Completable。
大家都知道RxJava切换线程使用subscribeOn指定被观察者的在哪个线程执行&＃xff0c;使用observeOn指定观察者在哪个线程执行&＃xff0c;通常我们写法如下:
subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
subscribeOn会返回一个ObservableSubscribeOn&＃xff0c;它是一个Observable&＃xff0c;根据前面介绍的订阅流程&＃xff0c;我们直接看ObservableSubscribeOn的subscribeActual操作&＃xff1a;
&＃64;Override
public void subscribeActual(final Observer observer) {//创建了内部的Observer&＃xff0c;其实这里类似上面介绍的Observable.create创建的发射器&＃xff0c;只不过发射器是Emitterfinal SubscribeOnObserver
}
创建了SubscribeOnObserver对象&＃xff0c;它是Observer类型的&＃xff0c;其实类似上面介绍的Observable.create创建的发射器&＃xff0c;只不过发射器是Emitter类型。接着给下游的observer添加订阅的监听&＃xff0c;最后是给SubscribeOnObserver设置disposable对象&＃xff0c;还记得在observable.create最后一步是给上游的ObservableOnSubscribe添加订阅吗&＃xff0c;那我们看看此处是如果给上游的observable添加订阅的&＃xff0c;首先scheduler是Schedulers.io()&＃xff0c;最终它是一个IoScheduler对象&＃xff0c;里面是通过CachedWorkerPool内部类创建了线程池&＃xff0c;创建线程池如下:
evictor &＃61; Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
而scheduler.scheduleDirect(new SubscribeTask(parent))中的SubscribeTask是一个Runnable&＃xff0c;所以最终通过线程池执行SubscribeTask的run方法&＃xff1a;
到了最后还是通过线程池执行Runnable来添加上游Observable的订阅&＃xff0c;并且把当前创建的SubscribeOnObserver传给了上游的observable&＃xff0c;这个跟我们上面介绍Observable.create中给上游的ObservableOnSubscribe添加订阅是一样的。
subscribeOn实际是创建了ObservableSubscribeOn的Observable&＃xff0c;它的订阅方法里面创建了SubscribeOnObserver&＃xff0c;通过线程池执行Runnable来达到上游Observable的订阅在子线程中执行&＃xff0c;这就是为什么subscribeOn能控制observable在哪个线程中执行的原因。同样如此observeOn也会有对应的observable&＃xff0c;它是ObservableObserveOn&＃xff0c;我们直接看它订阅的方法&＃xff1a;
同样如此&＃xff0c;可以看到先是拿到AndroidSchedulers中的worker&＃xff0c;它是HandlerWorker类型&＃xff0c;按道理说应该给下游的observer添加订阅监听啊&＃xff0c;怎么没有呢&＃xff0c;看官别急&＃xff0c;我们继续看ObserveOnObserver的订阅方法&＃xff1a;
我们的重点不在下游的observer订阅监听这&＃xff0c;在ObserveOnObserver的onNext方法中&＃xff0c;会调用schedule方法&＃xff0c;最终是通过HandlerWorker的schedule执行ObserveOnObserver&＃xff0c;因为ObserveOnObserver也是一个runnable实现类&＃xff0c;HandlerWorker中的schedule方法是通过主线程的Handler给主线程发送了一个Message&＃xff0c;所以我们回到ObserveOnObserver的run方法&＃xff0c;在run方法中会执行下游的onNext、onError等方法&＃xff0c;所以这就是为什么observeOn能让observer能在主线程中执行。
observeOn实际是创建了ObservableObserveOn的Observable&＃xff0c;它的订阅方法里面创建了ObserveOnObserver&＃xff0c;而ObserveOnObserver是实现了Runnable接口&＃xff0c;把它包装成message给主线程的Handler发送一条消息&＃xff0c;而ObserveOnObserver的run方法中会给下游的Observer发送数据。所以这就是observeOn能让observer在哪个线程中执行。
如果你理解了订阅的过程&＃xff0c;其实该问题很好理解&＃xff0c;subscribeOn是规定上游的observable在哪个线程中执行&＃xff0c;如果我们执行多次的subscribeOn的话&＃xff0c;从下游的observer到上游的observable的订阅过程&＃xff0c;最开始调用的subscribeOn返回的observable会把后面执行的subscribeOn返回的observable给覆盖了&＃xff0c;因此我们感官的是只有第一次的subscribeOn能生效。
那如何才能知道它实际在里面生效了呢&＃xff0c;我们可以通过doOnSubscribe来监听切实发生线程切换了。
上面分析了observeOn是指定下游的observer在哪个线程中执行&＃xff0c;所以这个更好理解&＃xff0c;看observeOn下一个observer是哪一个&＃xff0c;所以多次调用observeOn肯定是最后一个observeOn控制有效。
RxJava2.0相比于RxJava1.0
RxJava3.0相比与RxJava2.0
你以为完了吗&＃xff0c;还有背压没介绍呢&＃xff0c;好吧&＃xff0c;由于篇幅原因&＃xff0c;把背压放到下一篇单独来讲了。
介绍了发射器发送了onComplete事件后再不能发送onNext事件的原因&＃xff0c;介绍了map、flatMap以及concatMap的区别&＃xff0c;以及简单的介绍了其他的关键字。
介绍了Maybe、Single、Flowable、Completable几种观察者的区别&＃xff0c;以及他们在什么场景用。
介绍了RxJava切换线程通过subscribeOn控制上游的Observable订阅发生的线程&＃xff0c;observeOn控制下游的observer接收数据发生的线程&＃xff0c;以及为什么RxJava的subscribeOn只有第一次生效&＃xff0c;RxJava的observeOn多次调用离observer最近的一个才生效。
介绍了RxJava2.0相比于RxJava1.0有哪些变动&＃xff0c;RxJava3.0相比与RxJava2.0有哪些变动。
在这我分享一份自己收录整理的Android 核心知识笔记&＃43;架构视频&＃43;面试文档&＃xff0c;还有 高级架构技术进阶脑图、Android开发面试专题等一些高级进阶架构资料&＃xff0c;我平常在闲暇的时刻&＃xff0c;还会将这些精品资料拿出来反复翻阅。
总之是想通过学习资料和方法能够帮助大家学习提升进阶&＃xff0c;也节省大家在网上搜索资料的时间来学习&＃xff0c;也可以分享给身边好友一起学习&＃xff0c;如果你有需要的话&＃xff0c;可以私信我【666】或点击【Github地址】进行查看获取&＃xff01;&＃xff01;&＃xff01;
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有 