这里有新鲜出炉的精品教程,程序狗速度看过来!
.Net响应式编程框架Rx的Java版本开源实现,是一种JVM语言,有Groovy, Clojure, Scala and JRuby版本的实现
RxJava2相比RxJava1,它的改动还是很大的,那么下面这篇文章主要给大家总结了在RxJava 1升级到RxJava 2过程中踩过的一些“坑”,文中介绍的非常详细,对大家具有一定的参考学习价值,需要的朋友们下来要一起看看吧。
RxJava2介绍
RxJava2 发布已经有一段时间了,是对 RxJava 的一次重大的升级,由于我的一个库cv4j使用了 RxJava2 来尝鲜,但是 RxJava2 跟 RxJava1 是不能同时存在于一个项目中的,逼不得已我得把自己所有框架中使用 RxJava 的地方以及
App 中使用 RxJava 的地方都升级到最新版本。所以我整理并记录了一些已经填好的坑。分享出来供大家参考学习,下面来看看详细的介绍:
填坑记录
1. RxJava1 跟 RxJava2 不能共存
如果,在同一个module中同时使用RxJava1和RxJava2,类似如下:
- compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
- compile 'io.reactivex.rxjava2:rxjava:2.0.7'
- compile 'io.reactivex:rxandroid:1.2.0'
- compile 'io.reactivex:rxjava:1.1.5'
那么,很不幸你会遇到这样的错误
同理,在 App 中如果使用了 Rxjava2,但是某个第三方的 library 还在使用 Rxjava1 也会遇到同样的错误。
上面的错误是因为 RxAndroid 2.0.1 本身依赖了 RxJava 2.0.1。我们尝试去掉对 RxJava 的依赖,只留下 RxAndroid 。还是会遇到问题。
- compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
- //compile 'io.reactivex.rxjava2:rxjava:2.0.7'
- compile 'io.reactivex:rxandroid:1.2.0'
- //compile 'io.reactivex:rxjava:1.1.5'
所以使用RxAndroid不能去掉对RxJava的依赖,我是这样使用的。
- compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
- compile 'io.reactivex.rxjava2:rxjava:2.0.7'
官方也是这样解释的
- Because RxAndroid releases are few and far between,
- it is recommended you also explicitly depend on RxJava 's latest version for bug fixes and new features.'
最后,我建议要升级到 RxJava2 的时候必须所有使用的地方都要升级,并且用最新的版本。
2. 新增Flowable
RxJava1 中 Observable 不能很好地支持 backpressure ,会抛出MissingBackpressureException。所以在 RxJava2 中 Oberservable 不再支持 backpressure ,而使用新增的 Flowable 来支持 backpressure 。
Flowable的用法跟原先的Observable是一样的。
3. ActionN 和 FuncN 改名
ActionN 和 FuncN 遵循Java 8的命名规则。
其中,Action0 改名成Action,Action1改名成Consumer,而Action2改名成了BiConsumer,而Action3 - Action9都不再使用了,ActionN变成了
。
- Consumer<Object[]>
同样,Func改名成Function,Func2改名成BiFunction,Func3 - Func9 改名成 Function3 - Function9,FuncN 由
取代。
- Function<Object[], R>
4. Observable.OnSubscribe 变成 ObservableOnSubscribe
原先RxJava1的写法:
- Observable.create(new Observable.OnSubscribe<String>() {
- @Override
- public void call(Subscriber<? super String> subscriber) {
- subscriber.onNext("hello");
- }
- }).subscribe(new Action1<String>() {
- @Override
- public void call(String s) {
- System.out.println(s);
- }
- });
现在的写法:
- Observable.create(new ObservableOnSubscribe<String>() {
- @Override
- public void subscribe(ObservableEmitter<String> e) throws Exception {
- e.onNext("hello");
- }
- }).subscribe(new Consumer<String>() {
- @Override
- public void accept(String s) {
- System.out.println(s);
- }
- });
5. ObservableOnSubscribe 中使用 ObservableEmitter 发送数据给 Observer
结合上一条,ObservableOnSubscribe 不再使用 Subscriber 而是用 ObservableEmitter 替代。
ObservableEmitter 可以理解为发射器,是用来发出事件的,它可以发出三种类型的事件,通过调用emitter的
、
- onNext(T value)
和
- onComplete()
可以分别发出next事件、complete事件和error事件。 如果只关心next事件的话,只需单独使用
- onError(Throwable error)
即可。
- onNext()
需要特别注意,emitter的
调用后,Consumer不再接收任何next事件。
- onComplete()
6. Observable.Transformer 变成 ObservableTransformer
原先RxJava1的写法:
- /**
- * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
- * @param <T>
- * @return
- */
- public static <T> Observable.Transformer<T, T> toMain() {
- return new Observable.Transformer<T, T>() {
- @Override
- public Observable<T> call(Observable<T> tObservable) {
- return tObservable
- .subscribeOn(Schedulers.io())
- .observeOn(AndroidSchedulers.mainThread());
- }
- };
- }
现在的写法:
- /**
- * 跟compose()配合使用,比如ObservableUtils.wrap(obj).compose(toMain())
- * @param <T>
- * @return
- */
- public static < T > ObservableTransformer < T,
- T > toMain() {
- return new ObservableTransformer < T,
- T > () {
- @Override public ObservableSource < T > apply(Observable < T > upstream) {
- return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
- }
- };
- }
由于新增了Flowable,同理也增加了FlowableTransformer
- public static < T > FlowableTransformer < T,
- T > toMain() {
- return new FlowableTransformer < T,
- T > () {
- @Override public Publisher < T > apply(Flowable < T > upstream) {
- return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
- }
- };
- }
7. Subscription 改名为 Disposable
在 RxJava2 中,由于已经存在了
这个类,为了避免名字冲突将原先的
- org.reactivestreams.subscription
改名为
- rx.Subscription
。
- io.reactivex.disposables.Disposable
刚开始不知道,在升级 RxJava2 时发现
这个类完全没法做原先
- org.reactivestreams.subscription
的事情:(
- rx.Subscription
顺便说下,Disposable必须单次使用,用完就要销毁。
8. first() 用法改变
官方文档是这么描述的
的用法
- first()
1.x | 2.x |
---|---|
first() | RC3 renamed to firstElement and returns Maybe<T> |
first(Func1) | dropped, use filter(predicate).first() |
firstOrDefault(T) | renamed to first(T) and RC3 returns Single<T> |
firstOrDefault(Func1, T) | renamed to first(T) and RC3 returns Single<T> |
以
为例,
- first(Func1)
后面还使用了
- first(Func1)
,原先 Rxjava1会这样写
- push()
- ConnectableObservable<Data> connectableObservable = Observable
- .concat(Observable.from(list))
- .first(new Func1<Data, Boolean>() {
- @Override
- public Boolean call(Data data) {
- return DataUtils.isAvailable(data);
- }
- }).publish();
RxJava2 改成这样
- ConnectableObservable<Data> connectableObservable = Observable
- .concat(Observable.fromIterable(list))
- .filter(new Predicate<Data>() {
- @Override
- public boolean test(@NonNull Data data) throws Exception {
- return DataUtils.isAvailable(data);
- }
- }).firstElement().toObservable().publish();
9. toBlocking().y 被 blockingY() 取代
在我的框架中存在着一个Optional类,它跟Java 8的Optional作用差不多,原先是使用RxJava1来编写的。
- import rx.Observable;
- /**
- * 使用方法:
- * String s = null;
- * Optional.ofNullable(s).orElse("default")); // 如果s为null,则显示default,否则显示s的值
- * @author Tony Shen
- *
- */
- public class Optional < T > {
- Observable < T > obs;
- public Optional(Observable < T > obs) {
- this.obs = obs;
- }
- public static < T > Optional < T > of(T value) {
- if (value == null) {
- throw new NullPointerException();
- } else {
- return new Optional < T > (Observable.just(value));
- }
- }
- public static < T > Optional < T > ofNullable(T value) {
- if (value == null) {
- return new Optional < T > (Observable. < T > empty());
- } else {
- return new Optional < T > (Observable.just(value));
- }
- }
- public T get() {
- return obs.toBlocking().single();
- }
- public T orElse(T defaultValue) {
- return obs.defaultIfEmpty(defaultValue).toBlocking().single();
- }
- }
升级到RxJava2之后,
和
- get()
方法都会报错,修改之后是这样的。
- orElse()
- import io.reactivex.Observable;
- /**
- * 使用方法:
- * String s = null;
- * Optional.ofNullable(s).orElse("default"); // 如果s为null,则显示default,否则显示s的值
- * @author Tony Shen
- *
- */
- public class Optional < T > {
- Observable < T > obs;
- public Optional(Observable < T > obs) {
- this.obs = obs;
- }
- public static < T > Optional < T > of(T value) {
- if (value == null) {
- throw new NullPointerException();
- } else {
- return new Optional < T > (Observable.just(value));
- }
- }
- public static < T > Optional < T > ofNullable(T value) {
- if (value == null) {
- return new Optional < T > (Observable. < T > empty());
- } else {
- return new Optional < T > (Observable.just(value));
- }
- }
- public T get() {
- return obs.blockingSingle();
- }
- public T orElse(T defaultValue) {
- return obs.defaultIfEmpty(defaultValue).blockingSingle();
- }
- }
10. PublishSubject
包括 PublishSubject 以及各种 Subject(ReplaySubject、BehaviorSubject、AsyncSubject) 都不再支持backpressure。
总结
RxJava2 所带来的变化远远不止这些,以后遇到的话还会继续整理和总结,毕竟我使用的 RxJava2 还是很少的一部分内容。
RxJava2 最好到文档依然是官方文档。如果是新项目到话,可以毫不犹豫地使用RxJava2,如果是在线上已经成熟稳定的项目,可以再等等。对于新手的话,可以直接从 RxJava2 学起,RxJava1 就直接略过吧。对于老手,RxJava2 还是使用原来的思想,区别不大,从 RxJava1 迁移到 Rxjava2 也花不了多少工夫。
好了,以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流,谢谢大家对PHPERZ的支持。
来源: http://www.phperz.com/article/17/0906/337626.html