前言

• Rxjava由於其基於事件流的鏈式呼叫、邏輯簡潔 & 使用簡單的特點，深受各大 Android開發者的歡迎。

• 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需要注意的坑，希望大家會喜歡

1. 本系列文章主要基於 Rxjava 2.0
2. 接下來的時間，我將持續推出 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
  

  1. 而被觀察者的舊實現Observable
     不再支援 背壓Backpressure
     

3. 建立被觀察者（Observable） & 觀察者（Observer） 方式的區別

在`RxJava 2.0 `中，建立被觀察者（`Observable`） & 觀察者（Observer）的方式也與`RxJava 1.0 `有些區別：

• 對於建立被觀察者（Observable）

<-- RxJava 1.0 中 建立被觀察者 -->
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 中 建立被觀察者 -->
// 變化1：Observable.OnSubscribe介面名改成ObservableOnSubscribe 
Observable<Integer> observable=Observable.create(new ObservableOnSubscribe<Integer>() {

            // 變化2：複寫的call（Subscriber）改成 subscribe （ObservableEmitter）    
            // 注：引數也發生了變化，即Subscriber -> ObservableEmitter = 發射器
           @Override
            public void subscribe(ObservableEmitter<Integer> 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<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 介面（實現了Observer介面的抽象類）
// 與Observer介面的區別：對 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 中 建立觀察者（Observer） -->
        Observer<Integer> observer= new Observer<Integer>() {

            // 變化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

<-- 示例1 -->
   Disposable disposable = observable.subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                  //這裡接收資料項
            }
        }, 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. 這意味著，在這些方法裡呼叫會發生異常的方法不需要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<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（）操作符

主要變動在於：
1. RxJava 1.0實現的是：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);

1. RxJava 2.0 實現的是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 = 傳送資料，但區別在於訂閱後只能接受到1次
• 改動如下

<-- 原始碼分析 -->
// 變動1：Single被重新設計為 Reactive-Streams架構，即SingleSubscriber 改為：SingleObserver
interface SingleObserver<T> {
    // 變動2：多了一個回撥方法 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（）轉換成Single物件
// 即，Observable.just(1).toSingle()

6.3 更改Completable

• Completable的作用類似於 Observable = 傳送資料，但區別在於訂閱後只能接受 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()轉換成Completable物件
// 即，Observable.just(1).toCompletable()

7. 使用建議

對於學習 & 在專案中使用RxJava的版本選擇，我給出以下建議：

8. 總結

• 本文主要講解了RxJava 2.0相對於 RxJava 1.0的變動

• 從上面可以看到，RxJava 2.0相對於 RxJava 1.0最大的改動，主要是增加了被觀察者的新實現：Flowable & 用於背壓（Backpressure）的處理

• 接下來的時間，我將持續推出 Android中 Rxjava 2.0 的一系列文章，包括原理、操作符、應用場景、背壓等等 ，有興趣可以繼續關注Carson_Ho的安卓開發筆記！！

請點贊 / 評論幫頂！因為你的鼓勵是我寫作的最大動力！