前言

Rxjava由于其基于事件流的链式调用、逻辑简洁 & 使用简单的特点，深受各大 Android开发者的欢迎。

Github截图

如果还不了解RxJava，请看文章：Android：这是一篇清晰 & 易懂的Rxjava 入门教程

RxJava 2.0已于2016 - 10.29正式发布，对 RxJava 1.0进行了1次重大升级：实际使用的API 及方法有很大的区别

但 RxJava 2.0 的使用思路和 RxJava 1.0 非常类似

同时，由于RxJava 2.0 跟 RxJava 1.0 不能共存在1个项目中，所以假如你在使用RxJava 1.0需要升级到RxJava 2.0，则需要做一些转变
今天，我将为大家带来 RxJava 2.0 相对于RxJava 1.0 的升级总结 & 从RxJava 1.0升级到RxJava 2.0需要注意的坑，希望大家会喜欢

本系列文章主要基于 Rxjava 2.0

接下来的时间，我将持续推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操作符、应用场景、背压等等 ，有兴趣可以继续关注Carson_Ho的安卓开发笔记！！

示意图

1. 依赖包更改

由于RxJava 2.0 跟 RxJava 1.0 不能共存在1个项目中，所以依赖也不能共存，需要进行更换
改动如下

// 原本：`RxJava 1.0` 依赖compile 'io.reactivex:rxandroid:1.2.0'compile 'io.reactivex:rxjava:1.1.5'// 更改：`RxJava 2.0` 依赖compile 'io.reactivex.rxjava2:rxandroid:2.0.1'compile 'io.reactivex.rxjava2:rxjava:2.0.7'// 注：RxJava2 与 RxJava1 不能共存，即依赖不能同时存在

2. 增加被观察者的新实现：Flowable

由于 RxJava 1.0 中的被观察者Observable不能很好地支持背压（Backpressure）
所以，在 RxJava 2.0 中 增加了被观察者的新实现 Flowable 来支持背压Backpressure

而被观察者的旧实现Observable不再支持背压Backpressure

Flowable的使用与 Observable非常类似，关于使用具体请看文章：Android RxJava 背压策略：图文 + 实例全面解析

3. 创建被观察者（Observable） & 观察者（Observer）方式的区别

在RxJava 2.0中，创建被观察者（Observable） & 观察者（Observer）的方式也与RxJava 1.0有些区别：

对于创建被观察者（Observable）

<-- RxJava 1.0 中 创建被观察者 -->Observable observable = Observable.create(new Observable.OnSubscribe() {    @Override    public void call(Subscriber<? super String> subscriber) {        subscriber.onNext("Hello");        subscriber.onNext("Hi");        subscriber.onNext("Aloha");        subscriber.onCompleted();    }});<-- RxJava 2.0 中 创建被观察者 -->// 变化1：Observable.OnSubscribe接口名改成ObservableOnSubscribe Observable observable=Observable.create(new ObservableOnSubscribe() {            // 变化2：复写的call（Subscriber）改成 subscribe （ObservableEmitter）                // 注：参数也发生了变化，即Subscriber -> ObservableEmitter = 发射器           @Override            public void subscribe(ObservableEmitter e) throws Exception {                // 可发出三种类型的事件：next事件、complete事件&error事件                // 通过调用emitter.onNext(T value) 、onComplete()和onError(Throwable e)                e.onNext(1);                e.onNext(2);                e.onError(new Exception("发生错误了"));                e.onComplete();            }        });

对于创建观察者（Observer）

<-- RxJava 1.0 中 创建观察者（Observer） -->// 方法1：采用 Observer 接口Observer observer = new Observer() {    @Override    public void onNext(String s) {        Log.d(tag, "Item: " + s);    }    @Override    public void onCompleted() {        Log.d(tag, "Completed!");    }    @Override    public void onError(Throwable e) {        Log.d(tag, "Error!");    }};// 方法2：采用 Subscriber 接口（实现了Observer接口的抽象类）// 与Observer接口的区别：对 Observer接口进行了扩展：onStart()、unsubscribe()，但使用方式基本类似Subscriber subscriber = new Subscriber() {    @Override    public void onNext(String s) {        Log.d(tag, "Item: " + s);    }    @Override    public void onCompleted() {        Log.d(tag, "Completed!");    }    @Override    public void onError(Throwable e) {        Log.d(tag, "Error!");    }};<-- RxJava 2.0 中 创建观察者（Observer） -->        Observer observer= new Observer() {            // 变化1：增加回调方法onSubscribe()            // 作用：最先调用该方法，即适合做初始化工作            @Override            public void onSubscribe(Disposable d) {            // 传入的参数Disposable作用 类似于 Subsciption            // 即相当于订阅关系的开关，即可切断 观察者和被观察者的订阅关系            // 注：调用dispose() = 观察者无法接收事件，但被观察者还是会继续发送事件            }            @Override            public void onNext(Integer value) {            }            @Override            public void onError(Throwable e) {            }            // 变化2：onCompleted（）改成 onComplete（）            @Override            public void onComplete() {            }        }

4. 简化订阅方法

对于简化订阅的方式， RxJava 1 主要采用 ActionX接口 & FuncX接口
在 RxJava 2 中，主要是对这一系列接口的名字 按照Java8的命名规则 进行了修改，而使用方法不变

4.1 ActionX 和 FuncX 改名

对于 ActionX接口名的更改

RxJava 1	RxJava 2
Action0	Action
Action1	Consumer（接收1个参数）
Action2	BiConsumer （接收2个参数）
ActionN	Consumer （接收多个参数）
Action3 - Action9	不再使用

对于 FuncX接口名的更改

RxJava 1	RxJava 2
Func	Function （用于变换对象）
Func2	BiFunction
Func3 - Func9	Function3 - Function9
FuncN	Function

具体如下

示意图

示例

<-- 示例1 -->   Disposable disposable = observable.subscribe(new Consumer() {            @Override            public void accept(Integer integer) throws Exception {                  //这里接收数据项            }        }, new Consumer() {            @Override            public void accept(Throwable throwable) throws Exception {              //这里接收onError            }        }, new Action() {            @Override            public void run() throws Exception {              //这里接收onComplete。            }        });<-- 示例2 -->flowable.subscribe(        new Consumer() {//相当于onNext            @Override            public void accept(String s) throws Exception {            }        }, new Consumer() {//相当于onError            @Override            public void accept(Throwable throwable) throws Exception {            }        }, new Action() {//相当于onComplete，注意这里是Action            @Override            public void run() throws Exception {            }        }, new Consumer() {//相当于onSubscribe            @Override            public void accept(Subscription subscription) throws Exception {            }        });

4.2 RxJava2的接口方法都允许抛出异常

即，接口方法里加上了 throws Exception

// Action接口public interface Action {    void run() throws Exception;}// Consumer接口public interface Consumer {    void accept(T t) throws Exception;}// 注：  // 1. 这意味着，在这些方法里调用会发生异常的方法不需要try-catch  // 2. RxJava 2.0 不再支持 null 值，如果传入一个null会抛出 NullPointerException

5. 操作符的改变

对于操作符，RxJava 1.0与 RxJava 2.0 在命名 & 行为上大多数保持了一致
需要强调的是first（）、subscribeWith（）和 compose()操作符

5.1 first（）操作符

改动如下

RxJava 1.0	RxJava 2.0
first（）	改名为：firstElement（）
first(Func1)	弃用，改用为：filter(predicate).first()
firstOrDefault(T)	改名为：first（T）
firstOrDefault(Func1, T)	改名为：first（T）

示例

<-- RxJava 1.0 -->Observable          .concat(Observable.from(list))          .first(new Func1() {                @Override                public Boolean call(Data data) {                    return DataUtils.isAvailable(data);                }            }).publish();<-- RxJava 2.0 -->Observable          .concat(Observable.fromIterable(list))          .filter(new Predicate() {                @Override                public boolean test(@NonNull Data data) throws Exception {                    return DataUtils.isAvailable(data);                }            }).firstElement().toObservable().publish();

5.2 subscribeWith（）操作符

具体请看下图：

示意图

5.3 compose（）操作符

主要变动在于：

RxJava 1.0实现的是：rx.Observable.Transformer接口

继承自Func1, Observable>

<-- RxJava 1.0 中的用法 -->private static  Observable.Transformer createIOSchedulers() {        return new Observable.Transformer() {            @Override            public Observable call(Observable tObservable) {                return tObservable.subscribeOn(Schedulers.io())                        .unsubscribeOn(AndroidSchedulers.mainThread())                        .observeOn(AndroidSchedulers.mainThread());            }        };    }    public static  Observable.Transformer,T> applySchedulers() {        return createIOSchedulers（）;    }Action1 onNext = null;String[] items = { "item1", "item2", "item3" };Subscription subscription = Observable.from(items)                                      .compose(RxUtil.applySchedulers())                                      .map(new Func1() {                                                  @Override public Integer call(String s) {                                                      return Integer.valueOf(s);                                                  }                                              })                                      .subscribe(onNext);

RxJava 2.0 实现的是io.reactivex.ObservableTansformer

一个独立的接口

<-- RxJava 2.0 中的用法 -->public static  ObservableTransformer io2MainObservable() {        return new ObservableTransformer() {            @Override            public ObservableSource apply(Observable upstream) {                return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());            }        };    }    public static  ObservableTransformer applySchedulers() {        return io2MainObservable（）;    }Consumer onNext = null;String[] items = { "item1", "item2", "item3" };Disposable disposable = Observable.fromArray(items)                                  .compose(RxUtil.applySchedulers())                                  .map(new Function() {                                              @Override public Integer apply(String s) throws Exception {                                                  return Integer.valueOf(s);                                              }                                          })                                  .subscribe(onNext);

6. 额外

6.1 新增Processor

作用类似于 Subject & 继承自 Flowable = 支持背压控制

而Subject则不支持背压控制

使用如下

//Processor    AsyncProcessor processor = AsyncProcessor.create();    processor.subscribe(o -> Log.d("JG",o)); //three    processor.onNext("one");    processor.onNext("two");    processor.onNext("three");    processor.onComplete();//Subject    AsyncSubject subject = AsyncSubject.create();    subject.subscribe(o -> Log.d("JG",o));//three    subject.onNext("one");    subject.onNext("two");    subject.onNext("three");    subject.onComplete();

6.2 更改Single

Single的作用类似于 Observable = 发送数据，但区别在于订阅后只能接受到1次
改动如下

<-- 源码分析 -->// 变动1：Single被重新设计为 Reactive-Streams架构，即SingleSubscriber 改为：SingleObserverinterface SingleObserver {    // 变动2：多了一个回调方法 onSubscribe（）    void onSubscribe(Disposable d);     void onSuccess(T value);    void onError(Throwable error);}<-- 具体使用 -->Single single = Single.just(1l);single.subscribe(new SingleObserver() {    @Override    public void onSubscribe(Disposable d) {    }    @Override    public void onSuccess(Long value) {        // 和onNext是一样的    }    @Override    public void onError(Throwable e) {    }});// 注：普通Observable对象可通过toSingle（）转换成Single对象// 即，Observable.just(1).toSingle()

6.3 更改Completable

Completable的作用类似于 Observable = 发送数据，但区别在于订阅后只能接受 Complete 和 onError事件
改动如下

// 变动1：Completable被重新设计为 Reactive-Streams架构，即CompletableSubscriber 改为：CompletableObserverinterface CompletableObserver {    void onSubscribe(Disposable d);    void onComplete();    void onError(Throwable error);}<-- 具体使用 -->Completable Completable = Completable.just(1l);Completable.subscribe(new CompletableObserver() {    @Override    public void onSubscribe(Disposable d) {    }    @Override    public void onComplete(Long value) {    }    @Override    public void onError(Throwable e) {    }});// 注：普通Observable对象可通过toCompletable()转换成Completable对象// 即，Observable.just(1).toCompletable()

7. 使用建议

对于学习 & 在项目中使用RxJava的版本选择，我给出以下建议：

示意图

8. 总结

本文主要讲解了RxJava 2.0相对于 RxJava 1.0的变动
从上面可以看到，RxJava 2.0相对于 RxJava 1.0最大的改动，主要是增加了被观察者的新实现：Flowable & 用于背压（Backpressure）的处理

此处，我建议大家深入了解背压，请看文章：Android RxJava 背压策略：图文 + 实例全面解析

下面，我将继续深入讲解 Rxjava 的相关知识，感兴趣的同学可以继续关注本人运营的Wechat Public Account：
我想给你们介绍一个与众不同的Android微信公众号（福利回赠）
我想邀请您和我一起写Android（福利回赠）

请点赞！因为你的鼓励是我写作的最大动力！

相关文章阅读

操作符使用
Android：这是一篇清晰 & 易懂的Rxjava 入门教程
Android RxJava：最基础的操作符详解 - 创建操作符
Android RxJava：图文详解变换操作符
Android RxJava：组合 / 合并操作符详细教程
Android RxJava：功能性操作符全面讲解

实际应用讲解
Android RxJava 实际应用讲解：（无条件）网络请求轮询
Android RxJava 实际应用讲解：（有条件）网络请求轮询
Android RxJava 实际应用讲解：网络请求嵌套回调
Android RxJava 实际应用讲解：合并数据源
Android RxJava 实际应用讲解：从磁盘 / 内存缓存中获取缓存数据
Android RxJava 实际应用讲解：联合判断
Android RxJava：细说线程控制（切换 / 调度）（含Retrofit实例讲解）
Android RxJava 实际应用讲解：网络请求出错重连（结合Retrofit）

欢迎关注Carson_Ho的！

不定期分享关于安卓开发的干货，追求短、平、快，但却不缺深度。

Android(安卓)RxJava：2.0 相对于 1.0的更新 & 变化（含 RxJava 1.0的更新使用）

前言

目录

1. 依赖包更改

2. 增加被观察者的新实现：Flowable

3. 创建被观察者（Observable） & 观察者（Observer）方式的区别

4. 简化订阅方法

4.1 ActionX 和 FuncX 改名

4.2 RxJava2的接口方法都允许抛出异常

5. 操作符的改变

5.1 first（）操作符

5.2 subscribeWith（）操作符

5.3 compose（）操作符

6. 额外

6.1 新增Processor

6.2 更改Single

6.3 更改Completable

7. 使用建议

8. 总结

请点赞！因为你的鼓励是我写作的最大动力！

欢迎关注Carson_Ho的！

更多相关文章

随机推荐

Android(安卓)RxJava：2.0 相对于 1.0的更新 & 变化（含 RxJava 1.0的更新使用）

前言

目录

1. 依赖包更改

2. 增加被观察者的新实现：Flowable

3. 创建被观察者（Observable） & 观察者（Observer） 方式的区别

4. 简化订阅方法

4.1 ActionX 和 FuncX 改名

4.2 RxJava2的接口方法都允许抛出异常

5. 操作符的改变

5.1 first（）操作符

5.2 subscribeWith（）操作符

5.3 compose（）操作符

6. 额外

6.1 新增Processor

6.2 更改Single

6.3 更改Completable

7. 使用建议

8. 总结

请点赞！因为你的鼓励是我写作的最大动力！

欢迎关注Carson_Ho的！

更多相关文章

随机推荐

3. 创建被观察者（Observable） & 观察者（Observer）方式的区别