Rxjava2入門教程二：Observable與Observer響應式編程在Rxjava2中的典型實現

在RxJava中，函數響應式編程具體表現為一個觀察者（Observer）訂閱一個可觀察對象（Observable），通過創建可觀察對象發射數據流，經過一系列操作符（Operators）加工處理和線程調度器（Scheduler）在不同線程間的轉發，最后由觀察者接受並做出響應的一個過程
ObservableSource與Observer是RxJava2中最典型的一組觀察者與可觀察對象的組合，其他四組可以看做是這一組的改進版或者簡化版。

Observable

抽象類Observable是接口ObservableSource下的一個抽象實現，我們可以通過Observable創建一個可觀察對象發射數據流。

     Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> emitter) throws Exception { emitter.onNext("Hello World"); emitter.onComplete(); } }); 

調用Observable.create方法，創建一個可觀察對象，並通過onNext發送一條數據“Hello World”，然后通過onComplete發送完成通知。

Observer

創建一個觀察者Observer來接受並響應可觀察對象發射的數據

Observer<String> observer = new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String s) { System.out.println(s); } @Override public void onError(Throwable e) { e.printStackTrace(); } @Override public void onComplete() { System.out.println("接受完成"); } }; 

在onNext方法中接收到可觀察對象發射的數據"Hello World",並做出響應——打印到控制台。

Observer訂閱Observable

observable.subscribe(observer);

通過subscribe方法，使Observer與Observable建立訂閱關系，Observer與Observable便成為了一個整體，Observer便可對Observable中的行為作出響應。

Emitter/Observer

通過Observable.create創建可觀察對象時，我們可以發現具體執行發射動作的是由接口ObservableEmitter的實例化對象完成的，而ObservableEmitter<T> 繼承自 接口Emitter<T>，查看源碼接口Emitter的具體代碼如下：

public interface Emitter<T> { //用來發送數據，可多次調用，每調用一次發送一條數據 void onNext(@NonNull T value); //用來發送異常通知，只發送一次，若多次調用只發送第一條 void onError(@NonNull Throwable error); //用來發送完成通知，只發送一次，若多次調用只發送第一條 void onComplete(); } 

onNext：用來發送數據，可多次調用，每調用一次發送一條數據
onError：用來發送異常通知，只發送一次，若多次調用只發送第一條
onComplete：用來發送完成通知，只發送一次，若多次調用只發送第一條

onError與onComplete互斥，兩個方法只能調用一個不能同時調用

數據在發送時，出現異常可以調用onError發送異常通知也可以不調用，因為其所在的方法subscribe會拋出異常，

若數據在全部發送完之后均正常，可以調用onComplete發送一條完成通知

接口Observer中的三個方法（onNext,onError,onComplete）正好與Emitter中的三個方法相對應，對於Emitter中對應方法發送的數據或通知進行響應。

步驟簡化

去掉中間變量可以對之前的代碼簡化為以下形式：

    public void demo2() { Observable .create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> emitter) throws Exception { emitter.onNext("Hello World"); emitter.onComplete(); } }) .subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String s) { System.out.println(s); } @Override public void onError(Throwable e) { e.printStackTrace(); } @Override public void onComplete() { System.out.println("接受完成"); } }); } 

再應用Rxjava中強大的操作符，可以將代碼簡化成以下形式：

    public void demo3() { Observable.just("Hello World") .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { System.out.println(s); } }); } 

再通過λ表達式，可進一步簡化

    public void demo3_1() { Observable.just("Hello World").subscribe(System.out::println); } 

其中，just操作符可用來發送單條數據，數字，字符串，數組，對象，集合都可以當做單條數據發送。
Consumer可以看做是對觀察者Observer功能單一化之后的產物——消費者，上例中的Consumer通過其函數accept只接收可觀察對象發射的數據，不接收異常信息或完成信息。
如果想接收異常信息或完成信息可以用下面的代碼：

    public void demo4() { Observable.just("Hello World") .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { System.out.println(s); } }, new Consumer<Throwable>() { @Override public void accept(@NonNull Throwable throwable) throws Exception { throwable.printStackTrace(); } }, new Action() { @Override public void run() throws Exception { System.out.println("接受完成"); } }); } 

第二個參數Consumer規定泛型<Throwable>通過函數accept接收異常信息。
第三個參數Action也是對觀察者Observer功能單一化之后的產物--行動，通過函數run接收完成信息，作出響應行動。

發送數據序列

Observable可以發送單條數據也可以發送數據序列
通過最基礎的方法發送數據序列：

public void demo5(final List<String> list) { Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> emitter) throws Exception { try { for (String str : list) { emitter.onNext(str); } emitter.onComplete(); } catch (Exception e) { emitter.onError(e); } } }).subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { } @Override public void onNext(String s) { System.out.println(s); } @Override public void onError(Throwable e) { e.printStackTrace(); } @Override public void onComplete() { System.out.println("接受完成"); } }); } 

在subscribe方法中，遍歷集合list中的String元素，通過emitter.onNext(str)逐一發送；發送完成后通過emitter.onComplete()發送完成通知；如果發送過程中遇到異常，通過emitter.onError(e)發送異常信息。
Observer中通過onNext接收emitter發送的每一條信息並打印到控制台（emitter發送幾次，Observer便接收幾次），通過onError(Throwable e)接收異常信息，onComplete()接收完成信息。
同樣可以通過操作符對其進行簡化，如下;

    public void demo6(final List<String> list) { Observable .fromIterable(list) .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { System.out.println(s); } }); } 

再用λ表達式，進一步簡化

    public void demo6_1(final List<String> list) { Observable.fromIterable(list).subscribe(System.out::println); } 

其中fromIterable操作符，可用來將一個可迭代對象中的元素逐一發送

Disposable

在之前的例子中，可以看到Observer接口中還有一個方法

public void onSubscribe(Disposable d) { } 

是在觀察者Observer與可觀察對象Observable，建立訂閱關系后，回調這個方法，並且傳過來一個Disposable類型的參數，可通過Disposable來控制Observer與Observable之間的訂閱。
無論觀察者Observer以何種方式訂閱可觀察對象Observable，都會生成一個Disposable，如下：

public void demo7(final List<String> list) { Disposable disposable1 = Observable.just("Hello World") .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { System.out.println(s); } }); Disposable disposable2 = Observable .fromIterable(list) .subscribe(new Consumer<String>() { @Override public void accept(@NonNull String s) throws Exception { System.out.println(s); } }); } 

查看Disposable接口的源碼，如下：

public interface Disposable { void dispose(); boolean isDisposed(); } 

其中isDisposed()方法用來判斷當前訂閱是否失效，dispose()方法用來取消當前訂閱。
只有當觀察者Observer與可觀察對象Observable之間建立訂閱關系，並且訂閱關系有效時，Observer才能對Observable進行響應。如果Observer在響應Observable的過程中，訂閱關系被取消，則Observer無法對取消訂閱關系之后Observable的行為作出響應。
運行下面的代碼，當Observable接收到第5條數據時，取消訂閱關系。

public void demo8() { Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter<Integer> emitter) throws Exception { for (int i = 0; i < 10; i++) { System.out.println("發送" + i); emitter.onNext(i); } emitter.onComplete(); } }).subscribe(new Observer<Integer>() { private Disposable disposable; @Override public void onSubscribe(Disposable d) { disposable = d; } @Override public void onNext(Integer integer) { System.out.println("接收" + integer); if (integer > 4) disposable.dispose(); } @Override public void onError(Throwable e) { e.printStackTrace(); } @Override public void onComplete() { System.out.println("數據接受完成"); } }); } 

控制台日志如下：

 I/System.out: 發送0
 I/System.out: 接收0
 I/System.out: 發送1
 I/System.out: 接收1
 I/System.out: 發送2
 I/System.out: 接收2
 I/System.out: 發送3
 I/System.out: 接收3
 I/System.out: 發送4
 I/System.out: 接收4
 I/System.out: 發送5
 I/System.out: 接收5
 I/System.out: 發送6
 I/System.out: 發送7
 I/System.out: 發送8
 I/System.out: 發送9

可以發現取消訂閱關系之前，Observable發送一條數據，Observe便接收一條，但是取消訂閱關系之后，Observe將不再接收Observable發送的數據。