专栏首页Java与Android技术栈RxJava1 升级到 RxJava2 所踩过的坑

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 条评论
登录 后参与评论

相关文章

  • 给 Java 和 Android 构建一个简单的响应式Local Cache

    首先,Local Cache 不是类似于 Redis、Couchbase、Memcached 这样的分布式 Cache。Local Cache 适用于在单机环境...

    fengzhizi715
  • 常用的像素操作算法:图像加法、像素混合、提取图像中的ROI

    图像可以是看成是一个多维的数组。读取一张图片,可以看成是读入了一系列的像素内容。这些像素内容,按照不同的模式具有不同的格式。对于三通道的 RGB 位图来说,每个...

    fengzhizi715
  • Disruptor 实践:整合到现有的爬虫框架一. Disruptor二. 实践总结:

    NetDiscovery 是基于 Vert.x、RxJava 2 等框架实现的爬虫框架。

    fengzhizi715
  • 二十五、Hystrix累计统计流、分发流、最大并发流、配置流、功能流(附代码示例)

    上篇文章 介绍了Hystrix的“主流”:在滑动窗口内统计流、健康流。既然Hystrix的指标数据收集是基于事件驱动,那么自然可以多一些监听流,那么本文将做个收...

    YourBatman
  • Netty实现简单RPC调用

    我们知道Dubbo是一个RPC框架,那RPC框架需要实现什么?需要实现的是调用远程服务和本地服务一样方便,同时提高调用远程服务的性能。而服务端和客户端之间的关系...

    路行的亚洲
  • 基于 RxJava2+Retrofit2 精心打造的 Android 基础框架 XSnow

    XSnow ? 基于RxJava2+Retrofit2精心打造的Android基础框架,包含网络、上传、下载、缓存、事件总线、权限管理、数据库、图片加载、UI模...

    非著名程序员
  • 2.抽取代码(BaseActivity)

    六月的雨
  • RxAndroid从零开始学之五(常见操作符与三级缓存)

    RxAndroid的操作符有很多,本以为写了上一节的一些基本的Operator就可以正常编写代码了,但是后来在github上看googlesample,发现了一...

    Frank909
  • 一个简单的支持MySQL和SQLite3的DB接口

    simple_db.zip 相关联代码:https://github.com/eyjian/mooon/tree/master/common_library/...

    一见
  • Android中volley封装实践记录

    在项目中一般使用使用volley方式如下,用起来给人一种很乱的感觉,于是一种盘它的想法油然而生。

    砸漏

扫码关注云+社区

领取腾讯云代金券