前言

Rxjava由于其基于事件流的鏈式調(diào)用哼凯、邏輯簡潔 & 使用簡單的特點，深受各大 Android開發(fā)者的歡迎楚里。

如果還不了解RxJava断部，請看文章：Android：這是一篇清晰 & 易懂的Rxjava 入門教程

RxJava 2.0已于2016 - 10.29正式發(fā)布，對 RxJava 1.0進行了1次重大升級：實際使用的API 及方法有很大的區(qū)別班缎，但 RxJava 2.0 的使用思路和 RxJava 1.0 非常類似蝴光。同時，由于RxJava 2.0 跟 RxJava 1.0 不能共存在1個項目中达址，所以假如你在使用RxJava 1.0需要升級到RxJava 2.0蔑祟，則需要做一些轉(zhuǎn)變
今天，我將為大家?guī)?RxJava 2.0 相對于RxJava 1.0 的升級總結 & 從RxJava 1.0升級到RxJava 2.0需要注意的坑沉唠，希望大家會喜歡

Carson帶你學RxJava系列文章疆虚，包括 原理、操作符满葛、應用場景装蓬、背壓等等，請關注看文章：Android：這是一份全面 & 詳細的RxJava學習指南

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. 增加被觀察者的新實現(xiàn)：Flowable

由于 RxJava 1.0 中的被觀察者Observable不能很好地支持背壓（Backpressure）
所以乳蛾，在 RxJava 2.0 中 增加了被觀察者的新實現(xiàn) Flowable 來支持背壓Backpressure

而被觀察者的舊實現(xiàn)Observable不再支持背壓Backpressure

Flowable的使用與 Observable非常類似暗赶，關于使用具體請看文章：Android RxJava 背壓策略：圖文 + 實例全面解析

3. 創(chuàng)建被觀察者（Observable） & 觀察者（Observer）方式的區(qū)別

在RxJava 2.0中，創(chuàng)建被觀察者（Observable） & 觀察者（Observer）的方式也與RxJava 1.0有些區(qū)別：

對于創(chuàng)建被觀察者（Observable）

<-- RxJava 1.0 中 創(chuàng)建被觀察者 -->
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("Hi");
        subscriber.onNext("Aloha");
        subscriber.onCompleted();
    }
});

<-- RxJava 2.0 中 創(chuàng)建被觀察者 -->
// 變化1：Observable.OnSubscribe接口名改成ObservableOnSubscribe 
Observable<Integer> observable=Observable.create(new ObservableOnSubscribe<Integer>() {

            // 變化2：復寫的call（Subscriber）改成 subscribe （ObservableEmitter）    
            // 注：參數(shù)也發(fā)生了變化肃叶，即Subscriber -> ObservableEmitter = 發(fā)射器
           @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                // 可發(fā)出三種類型的事件：next事件蹂随、complete事件&error事件
                // 通過調(diào)用emitter.onNext(T value) 、onComplete()和onError(Throwable e)
                e.onNext(1);
                e.onNext(2);
                e.onError(new Exception("發(fā)生錯誤了"));
                e.onComplete();
            }
        });

對于創(chuàng)建觀察者（Observer）

<-- RxJava 1.0 中 創(chuàng)建觀察者（Observer） -->
// 方法1：采用 Observer 接口
Observer<String> observer = new Observer<String>() {
    @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 接口（實現(xiàn)了Observer接口的抽象類）
// 與Observer接口的區(qū)別：對 Observer接口進行了擴展：onStart()因惭、unsubscribe()岳锁，但使用方式基本類似

Subscriber<String> subscriber = new Subscriber<String>() {
    @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 中 創(chuàng)建觀察者（Observer） -->
        Observer<Integer> observer= new Observer<Integer>() {

            // 變化1：增加回調(diào)方法onSubscribe()
            // 作用：最先調(diào)用該方法，即適合做初始化工作
            @Override
            public void onSubscribe(Disposable d) {
            // 傳入的參數(shù)Disposable作用 類似于 Subsciption
            // 即相當于訂閱關系的開關蹦魔，即可切斷 觀察者和被觀察者的訂閱關系
            // 注：調(diào)用dispose() = 觀察者無法接收事件激率，但被觀察者還是會繼續(xù)發(fā)送事件
            }

            @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的命名規(guī)則 進行了修改乒躺，而使用方法不變

4.1 ActionX 和 FuncX 改名

對于 ActionX接口名的更改

RxJava 1	RxJava 2
Action0	Action
Action1	Consumer（接收1個參數(shù)）
Action2	BiConsumer （接收2個參數(shù)）
ActionN	Consumer<Object[]> （接收多個參數(shù)）
Action3 - Action9	不再使用

對于 FuncX接口名的更改

RxJava 1	RxJava 2
Func	Function （用于變換對象）
Func2	BiFunction
Func3 - Func9	Function3 - Function9
FuncN	Function<Object[], R>

具體如下

示意圖

示例

<-- 示例1 -->
   Disposable disposable = observable.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                  //這里接收數(shù)據(jù)項
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
              //這里接收onError
            }
        }, new Action() {
            @Override
            public void run() throws Exception {
              //這里接收onComplete招盲。
            }
        });

<-- 示例2 -->
flowable.subscribe(
        new Consumer<String>() {//相當于onNext
            @Override
            public void accept(String s) throws Exception {
            }
        }, new Consumer<Throwable>() {//相當于onError
            @Override
            public void accept(Throwable throwable) throws Exception {
            }
        }, new Action() {//相當于onComplete，注意這里是Action
            @Override
            public void run() throws Exception {
            }
        }, new Consumer<Subscription>() {//相當于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<T> {
    void accept(T t) throws Exception;
}

// 注：
  // 1. 這意味著曹货，在這些方法里調(diào)用會發(fā)生異常的方法不需要try-catch
  // 2. RxJava 2.0 不再支持 null 值，如果傳入一個null會拋出 NullPointerException

5. 操作符的改變

對于操作符讳推，RxJava 1.0與 RxJava 2.0 在命名 & 行為上大多數(shù)保持了一致
需要強調(diào)的是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<Data, Boolean>() {
                @Override
                public Boolean call(Data data) {
                    return DataUtils.isAvailable(data);
                }
            }).publish();

<-- RxJava 2.0 -->
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();

5.2 subscribeWith（）操作符

具體請看下圖：

示意圖

5.3 compose（）操作符

主要變動在于：

RxJava 1.0實現(xiàn)的是：rx.Observable.Transformer接口

繼承自Func1<Observable<T>, Observable<R>>

<-- RxJava 1.0 中的用法 -->
private static <T> Observable.Transformer<T, T> createIOSchedulers() {
        return new Observable.Transformer<T, T>() {
            @Override
            public Observable<T> call(Observable<T> tObservable) {
                return tObservable.subscribeOn(Schedulers.io())
                        .unsubscribeOn(AndroidSchedulers.mainThread())
                        .observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> Observable.Transformer<JsonResult<T>,T> applySchedulers() {
        return createIOSchedulers（）;
    }

Action1<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Subscription subscription = Observable.from(items)
                                      .compose(RxUtil.<String>applySchedulers())
                                      .map(new Func1<String, Integer>() {
                                                  @Override public Integer call(String s) {
                                                      return Integer.valueOf(s);
                                                  }
                                              })
                                      .subscribe(onNext);

RxJava 2.0 實現(xiàn)的是io.reactivex.ObservableTansformer<Upstream, Downstream>

一個獨立的接口

<-- RxJava 2.0 中的用法 -->
public static <T> ObservableTransformer<T, T> io2MainObservable() {
        return new ObservableTransformer<T, T>() {
            @Override
            public ObservableSource<T> apply(Observable<T> upstream) {
                return upstream.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> ObservableTransformer<T, T> applySchedulers() {
        return io2MainObservable（）;
    }

Consumer<Integer> onNext = null;
String[] items = { "item1", "item2", "item3" };
Disposable disposable = Observable.fromArray(items)
                                  .compose(RxUtil.<String>applySchedulers())
                                  .map(new Function<String, Integer>() {
                                              @Override public Integer apply(String s) throws Exception {
                                                  return Integer.valueOf(s);
                                              }
                                          })
                                  .subscribe(onNext);

6. 額外

6.1 新增Processor

作用類似于 Subject & 繼承自 Flowable = 支持背壓控制

而Subject則不支持背壓控制

使用如下

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

//Subject
    AsyncSubject<String> 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 = 發(fā)送數(shù)據(jù)银觅，但區(qū)別在于訂閱后只能接受到1次
改動如下

<-- 源碼分析 -->
// 變動1：Single被重新設計為 Reactive-Streams架構蜕衡，即SingleSubscriber 改為：SingleObserver
interface SingleObserver<T> {
    // 變動2：多了一個回調(diào)方法 onSubscribe（）
    void onSubscribe(Disposable d); 
    void onSuccess(T value);
    void onError(Throwable error);
}

<-- 具體使用 -->
Single<Long> single = Single.just(1l);

single.subscribe(new SingleObserver<Long>() {
    @Override
    public void onSubscribe(Disposable d) {
    }

    @Override
    public void onSuccess(Long value) {
        // 和onNext是一樣的
    }

    @Override
    public void onError(Throwable e) {
    }
});

// 注：普通Observable對象可通過toSingle（）轉(zhuǎn)換成Single對象
// 即，Observable.just(1).toSingle()

6.3 更改Completable

Completable的作用類似于 Observable = 發(fā)送數(shù)據(jù)设拟，但區(qū)別在于訂閱后只能接受 Complete 和 onError事件
改動如下

// 變動1：Completable被重新設計為 Reactive-Streams架構，即CompletableSubscriber 改為：CompletableObserver
interface CompletableObserver<T> {
    void onSubscribe(Disposable d);
    void onComplete();
    void onError(Throwable error);
}

<-- 具體使用 -->
Completable<Long> Completable = Completable.just(1l);
Completable.subscribe(new CompletableObserver<Long>() {
    @Override
    public void onSubscribe(Disposable d) {
    }

    @Override
    public void onComplete(Long value) {

    }

    @Override
    public void onError(Throwable e) {
    }
});

// 注：普通Observable對象可通過toCompletable()轉(zhuǎn)換成Completable對象
// 即久脯，Observable.just(1).toCompletable()

7. 使用建議

對于學習 & 在項目中使用RxJava的版本選擇纳胧，我給出以下建議：

示意圖

8. 總結

本文主要講解了RxJava 2.0相對于 RxJava 1.0的變動
Carson帶你學RxJava系列文章：

入門
Carson帶你學Android：這是一篇清晰易懂的Rxjava入門教程
 Carson帶你學Android：面向初學者的RxJava使用指南
 Carson帶你學Android：RxJava2.0到底更新了什么？
原理
Carson帶你學Android：圖文解析RxJava原理
 Carson帶你學Android：手把手帶你源碼分析RxJava
使用教程：操作符
Carson帶你學Android：RxJava操作符教程
 Carson帶你學Android：RxJava創(chuàng)建操作符
 Carson帶你學Android：RxJava功能性操作符
 Carson帶你學Android：RxJava過濾操作符
 Carson帶你學Android：RxJava組合/合并操作符
 Carson帶你學Android：RxJava變換操作符
 Carson帶你學Android：RxJava條件/布爾操作符
實戰(zhàn)
Carson帶你學Android：什么時候應該使用Rxjava帘撰？(開發(fā)場景匯總)
Carson帶你學Android：RxJava線程控制(含實例講解)
Carson帶你學Android：圖文詳解RxJava背壓策略
 Carson帶你學Android：RxJava跑慕、Retrofit聯(lián)合使用匯總(含實例教程)
Carson帶你學Android：優(yōu)雅實現(xiàn)網(wǎng)絡請求嵌套回調(diào)
Carson帶你學Android：網(wǎng)絡請求輪詢(有條件)
Carson帶你學Android：網(wǎng)絡請求輪詢(無條件)
Carson帶你學Android：網(wǎng)絡請求出錯重連(結合Retrofit)
Carson帶你學Android：合并數(shù)據(jù)源
 Carson帶你學Android：聯(lián)想搜索優(yōu)化
 Carson帶你學Android：功能防抖
 Carson帶你學Android：從磁盤/內(nèi)存緩存中獲取緩存數(shù)據(jù)
Carson帶你學Android：聯(lián)合判斷

歡迎關注Carson_Ho的簡書

不定期分享關于安卓開發(fā)的干貨，追求短摧找、平核行、快，但卻不缺深度蹬耘。

Carson帶你學Android：RxJava2.0到底更新了什么？

Carson帶你學Android：RxJava2.0到底更新了什么？

前言

目錄

1. 依賴包更改

2. 增加被觀察者的新實現(xiàn)：Flowable

3. 創(chuàng)建被觀察者（Observable） & 觀察者（Observer）方式的區(qū)別

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的簡書

請點贊芝雪！因為你的鼓勵是我寫作的最大動力！

Carson帶你學Android：RxJava2.0到底更新了什么？

前言

目錄

1. 依賴包更改

2. 增加被觀察者的新實現(xiàn)：Flowable

3. 創(chuàng)建被觀察者（Observable） & 觀察者（Observer） 方式的區(qū)別

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. 創(chuàng)建被觀察者（Observable） & 觀察者（Observer）方式的區(qū)別