RxJava1 升级到 RxJava2 所踩过的坑

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'

那么,很不幸你会遇到这样的错误

Rxjava1和Rxjava2无法共存.jpeg

同理,在 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'

去掉对Rxjava的依赖.jpeg

所以使用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 中 Observable 不再支持 backpressure ,而使用新增的 Flowable 来支持 backpressure 。关于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()和onError(Throwable error)可以分别发出next事件、complete事件和error事件。 如果只关心next事件的话,只需单独使用onNext()即可。

需要特别注意,emitter的onComplete()调用后,Consumer不再接收任何next事件。

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)后面还使用了push(),原先 Rxjava1会这样写

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 也花不了多少工夫。

参考资料:

  1. https://github.com/ReactiveX/RxJava/wiki
  2. https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0
  3. http://www.dundunwen.com/article/275b1d92-f9da-4bb8-b111-3aa8a6ace245.html

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

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏BaronTalk

RxJava系列六(从微观角度解读RxJava源码)

前言 通过前面五个篇幅的介绍,相信大家对RxJava的基本使用以及操作符应该有了一定的认识。但是知其然还要知其所以然;所以从这一章开始我们聊聊源码,分析RxJ...

4157
来自专栏WindCoder

进击微信小程序-Java版AES/CBC/PKCS7Padding加解密字符串

进击微信小程序,最初找到腾讯云给的解决方案,无奈后端是node.js,而且才一开始配上去还不能用,之后仔细看readme,摸到它开源的Wafer,本来以为找到终...

2672
来自专栏杨逸轩 ' sBlog

Java实现微信跳一跳抓包修改分数

3826
来自专栏三流程序员的挣扎

RxJava 连接操作符

看注释意思是将所有数据按原来的顺序缓存起来,就是不知道观察者什么时候订阅,什么时候解除订阅,所以缓存起来,以后直接用。

2122
来自专栏Java与Android技术栈

Cold Observable 和 Hot Observable

Hot Observable 无论有没有 Subscriber 订阅,事件始终都会发生。当 Hot Observable 有多个订阅者时,Hot Observa...

1602
来自专栏androidBlog

Rxjava 2.x 源码系列 - 线程切换 (上)

版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/gdutxiaoxu/article/de...

3273
来自专栏freesan44

三重Des对称加密在Android、Ios 和Java 平台的实现

如今手机app五彩缤纷,确保手机用户的数据安全是开发人员必须掌握的技巧,下面通过实例介绍DES在android、ios、java平台的使用方法;

1161
来自专栏三流程序员的挣扎

RxJava 错误处理操作符

在源 Observable 遇到错误时,立即停止源 Observable 的数据发送,并用新的 Observable 对象进行新的数据发送。

3032
来自专栏ccylovehs

武电实业卡密接口对接经验分享

以往卡密都是电信直接提供txt文件我们这边再导入数据库中,现在电信要求厂家获取卡密必须通过接口调用,此次对接主要用到三个方面知识:

1091
来自专栏Python疯子

python hashlib模块

hashlib主要提供字符加密功能,将md5和sha模块整合到了一起,支持md5,sha1, sha224, sha256, sha384, sha512等算法

1002

扫码关注云+社区

领取腾讯云代金券