使用了 RxJava2 有一段時間了，深深感受到了其“牛逼”之處。下面，就從 RxJava2 的基礎開始，一步步與大家分享一下這個強大的非同步庫的用法！RxJava 是 一個在 Java VM 上使用可觀測的序列來組成非同步的、基於事件的程式的庫，也就是用於實現非同步操作的庫。

1.RxJava2 基礎

RxJava可以濃縮為非同步兩個字，其核心的東西不外乎兩個， Observables（被觀察者） 和 Observable（觀察者）。Observables可以發出一系列的 事件（例如網路請求、複雜計算、資料庫操作、檔案讀取等），事件執行結束後交給Observable 的回撥處理。

1.RxJava2 的觀察者模式

觀察者模式是物件的行為模式，也叫做釋出-訂閱(Publish/Subscribe)模式、模型-檢視(Model/View)模式、源-監聽器(Source/Listener)模式或從屬者(Dependents)模式。

什麼是觀察者模式？舉個栗子，Android中View的點選監聽器的實現，View是被觀察者，OnClickListener物件是觀察者，Activity要如何知道View被點選了？那就是派一個OnClickListener物件，入駐View，與View達成一個訂閱關係，一旦View被點選了，就通過OnClickListener物件的OnClick方法傳達給Activity。採用觀察者模式可以避免去輪詢檢查，節約有限的cpu資源。

RxJava 作為一個工具庫，使用的便是通用形式的觀察者模式：

普通事件：onNext()，相當於 onClick()、onEvent()；特殊事件：onCompleted() 和 onError()

如圖所示，RxJava 的基本概念分別為：Observable(被觀察者，事件源)，Observer(觀察者，訂閱者)，subscribe (訂閱)、事件；不同的是，RxJava 把多個事件看做一個佇列，並對每個事件單獨處理。在一個佇列中 onCompleted() 和 onError()，只有一個會被呼叫。如果呼叫了 onCompleted() 就說明佇列執行完畢，沒有出現異常，否則呼叫 onError() 方法並終止佇列。

2.RxJava2 響應式程式設計結構

什麼是響應式程式設計？舉個栗子，a = b + c; 這句程式碼將b+c的值賦給a，而之後如果b和c的值改變了不會影響到a，然而，對於響應式程式設計，之後b和c的值的改變也動態影響著a，意味著a會隨著b和c的變化而變化。

響應式程式設計的組成為Observable/Operator/Subscriber，RxJava在響應式程式設計中的基本流程如下：

這個流程，可以簡單的理解為：Observable -> Operator1 -> Operator2 -> Operator3 -> Subscriber

1. Observable發出一系列事件，他是事件的產生者；
2. Subscriber負責處理事件，他是事件的消費者；
3. Operator是對Observable發出的事件進行修改和變換；
4. 若事件從產生到消費不需要其他處理，則可以省略掉中間的Operator，從而流程變為Obsevable -> Subscriber；
5. Subscriber通常在主執行緒執行，所以原則上不要去處理太多的事務，而這些複雜的處理則交給Operator；
   3.建立一個完整的 RxJava2 呼叫

首先需要新增 RxJava2 在 Android 中的 Gradle 依賴：

compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
compile "io.reactivex.rxjava2:rxjava:2.0.8"

RxJava2 可以通過下面這幾種方法建立被觀察者：

// 傳送對應的方法
Observable.create(new ObservableOnSubscribe<String>() {
    // 預設在主執行緒裡執行該方法
    @Override
    public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
        e.onNext("Hello");
        e.onNext("World");
        // 結束標識
        e.onComplete();
    }
});
// 傳送多個數據
Observable.just("Hello", "World");
// 傳送陣列
Observable.fromArray("Hello", "World");
// 傳送一個數據
Observable.fromCallable(new Callable<String>() {
    @Override
    public String call() throws Exception {
        return "Hello";
    }
});

RxJava2 支援鏈式程式設計，下來我們建立被觀察者，然後建立觀察者並訂閱：

// 建立被觀察者
Observable.just("Hello", "World")
// 將被觀察者切換到子執行緒
.subscribeOn(Schedulers.io())
// 將觀察者切換到主執行緒
.observeOn(AndroidSchedulers.mainThread())
// 建立觀察者並訂閱
.subscribe(new Observer<String>() {
    // Disposable 相當於RxJava1.x中的 Subscription，用於解除訂閱
    private Disposable disposable;
    @Override
    public void onSubscribe(Disposable d) {
        disposable = d;
    }
    @Override
    public void onNext(String s) {
        Log.i("JAVA", "被觀察者向觀察者傳送的資料:" + s);
        if (s == "-1") {   // "-1" 時為異常資料，解除訂閱
            disposable.dispose();
        }
    }
    @Override
    public void onError(Throwable e) {
    }
    @Override
    public void onComplete() {
    }
});

一旦 Observer 訂閱了 Observable，Observable 就會呼叫 Observer 的 onNext()、onCompleted()、onError() 等方法。至此一個完整的 RxJava 呼叫就完成了。看一下輸出的Log：

I/JAVA: 被觀察者向觀察者傳送的資料:Hello

I/JAVA: 被觀察者向觀察者傳送的資料:World

若喜歡簡潔、定製服務，那麼可以實現的方法跟上面的實現方法是對應起來的，大家看引數就知道哪個對應哪個了，你可以通過new Consumer（不需要實現的方法你可以不寫，看上去更簡潔），Consumer就是消費者的意思，可以理解為消費了 onNext 等事件：

Observable.just("Hello", "World")
.subscribe(new Consumer<String>() {
    @Override
    public void accept(@NonNull String s) throws Exception {
        Log.i("JAVA", "被觀察者向觀察者傳送的資料:" + s);
    }
}, new Consumer<Throwable>() {
    @Override
    public void accept(@NonNull Throwable throwable) throws Exception {
    }
}, new Action() {
    @Override
    public void run() throws Exception {
    }
}, new Consumer<Disposable>() {
    @Override
    public void accept(@NonNull Disposable disposable) throws Exception {
    }
});

4.RxJava2 的操作符

RxJava中提供了大量不同種類，不同場景的Operators(操作符)，RxJava的強大性就來自於它所定義的操作符。主要分類：

其中有一些高頻使用的操作符如下：

5.RxJava2 執行緒排程器

排程器 Scheduler 用於控制操作符和被觀察者事件所執行的執行緒，不同的排程器對應不同的執行緒。RxJava提供了5種排程器：

可以使用 subscribeOn() 和 ObserveOn() 操作符進行執行緒排程，讓 Observable 在一個特定的排程器上執行。subscribeOn() 指定 subscribe() 所發生的執行緒，事件產生的執行緒。ObserveOn() 指定 Observer 所執行在的執行緒，事件消費的執行緒。

6.RxJava2 模擬傳送驗證碼倒計時功能：

public void onCodeClick() {
    final long count = 60; // 設定60秒
    Observable.interval(0, 1, TimeUnit.SECONDS)
            .take(count + 1)
            .map(new Function<Long, Long>() {
                @Override
                public Long apply(@NonNull Long aLong) throws Exception {
                    return count - aLong; // 由於是倒計時，需要將倒計時的數字反過來
                }
            })
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(new Consumer<Disposable>() {
                @Override
                public void accept(@NonNull Disposable disposable) throws Exception {
                    button.setEnabled(false);
                    button.setTextColor(Color.GRAY);
                }
            })
            .subscribe(new Observer<Long>() {
                @Override
                public void onSubscribe(Disposable d) {
                }
                @Override
                public void onNext(Long aLong) {
                    button.setText(aLong + "秒後重發");
                }
                @Override
                public void onError(Throwable e) {
                }
                @Override
                public void onComplete() {
                    button.setEnabled(true);
                    button.setTextColor(Color.RED);
                    button.setText("傳送驗證碼");
                }
            });
}

2.RxJava2 系列框架

3.RxJava2 與 Retrofit 的使用

RxJava 與 Retrofit 的使用，更像我們的 AsyncTask，通過網路獲取資料然後通過 Handler 更新UI。首先需要匯入依賴：

compile 'com.squareup.retrofit2:retrofit:2.2.0'
compile 'com.squareup.retrofit2:converter-gson:2.2.0'
compile 'com.squareup.retrofit2:adapter-rxjava2:2.2.0'

1.模擬使用者登陸獲取使用者資料

1.Bean物件：

public class UserParam {
    private String param1;
    private String param2;
    public UserParam(String param1, String param2) {
        this.param1 = param1;
        this.param2 = param2;
    }
    // 省略了 getter setter
}
public class NetBean {
    private FormBean form;
    // 省略了 getter setter
    public static class FormBean {
        private String username;
        private String password;
        // 省略了 getter setter
    }
}
public class UserBean {
    private String username;
    private String password;
    public UserBean(String username, String password) {
        this.username = username;
        this.password = password;
    }
    // 省略了 getter setter
}

2.ApiService，這裡返回Observable物件，也就是我們RxJava的被觀察者

public interface ApiService {
    @FormUrlEncoded
    @POST("/post")
    Observable<NetBean> getUserInfo(@Field("username")String username,
                                    @Field("password")String password);
}

3.RxJava + Retrofit 的實現

// 構建Retrofit
ApiService apiService = new Retrofit.Builder()
        .baseUrl("http://httpbin.org/")
        .addConverterFactory(GsonConverterFactory.create()) // RxJava2與Gson混用
        .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // RxJava2與Retrofit混用
        .build()
        .create(ApiService.class);

// 構建RxJava
UserParam param = new UserParam("zhangsan", "123");
// 傳送param引數
Observable.just(param)
        // flatMap方法是用於資料格式轉換的方法，引數一表示原資料，
        // 引數二表示轉換的資料，那麼就是通過傳送網路引數，轉換成網路返回的資料，呼叫Retrofit
        .flatMap(new Function<UserParam, ObservableSource<NetBean>>() {
            @Override
            public ObservableSource<NetBean> apply(@NonNull UserParam userParam)
                    throws Exception {
                // 1.傳送網路請求，獲取NetBean
                return apiService.getUserInfo(userParam.getParam1(), userParam.getParam2());
            }
        })
        .flatMap(new Function<NetBean, ObservableSource<UserBean>>() {
            @Override
            public ObservableSource<UserBean> apply(@NonNull NetBean netBean)
                    throws Exception {
                UserBean user = new UserBean(netBean.getForm().getUsername(),
                        netBean.getForm().getPassword());
                // 2.轉換NetBean資料為我們需要的UserBean資料
                return Observable.just(user);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<UserBean>() {
            @Override
            public void accept(@NonNull UserBean userBean) throws Exception {
                Log.i("JAVA", "" + "使用者名稱:" + userBean.getUsername()
                        + ", 密碼:" + userBean.getPassword());
            }
        });

2.模擬合併本地與伺服器購物車列表

這個案例其實就是使用者新增購物車的時候，首先會在本地儲存一份，然後發現如果沒有網路，那麼沒辦法提交到伺服器上，只能等下一次有網路的時候採用本地資料庫和伺服器資料的合併來實現上傳到伺服器。

首先需要準備 Retrofit 物件和獲取本地資料、網路資料的方法：

private ApiService apiService;
@Override
protected void onCreate(@Nullable Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    // 省略
    // 構建Retrofit
    apiService = new Retrofit.Builder()
            .baseUrl("http://httpbin.org/")
            .addConverterFactory(GsonConverterFactory.create()) // RxJava2與Gson混用
            .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // RxJava2與Retrofit混用
            .build()
            .create(ApiService.class);
}
/**
 * 獲取本地資料
 */
private Observable<List<String>> getDataForLocal() {
    List<String> list = new ArrayList<>();
    list.add("購物車的商品1");
    list.add("購物車的商品2");
    return Observable.just(list);
}
/**
 * 獲取網路資料
 */
private Observable<List<String>> getDataForNet() {
    return Observable.just("shopName")
        // flatMap方法是用於資料格式轉換的方法，引數一表示原資料，
        // 引數二表示轉換的資料，那麼就是通過傳送網路引數，轉換成網路返回的資料，呼叫Retrofit
        .flatMap(new Function<String, ObservableSource<NetBean>>() {
            @Override
            public ObservableSource<NetBean> apply(@NonNull String s) throws Exception {
                // 1.傳送網路請求，獲取資料
                return apiService.getCartList(s);
            }
        }).flatMap(new Function<NetBean, ObservableSource<List<String>>>() {
            @Override
            public ObservableSource<List<String>> apply(@NonNull NetBean netBean) throws Exception {
                // String shop = netBean.get_$Args257().getShopName();
                String shop = "購物車的商品3";
                List<String> list = new ArrayList<>();
                list.add(shop);
                // 2.轉換NetBean資料為我們需要的List<String>資料
                return Observable.just(list);
            }
        }).subscribeOn(Schedulers.io());
}

然後就可以建立被觀察者並訂閱了，來完成合並本地與伺服器購物車列表操作：

// merge操作符: 將兩個ObservableSource合併為一個ObservableSource
Observable.merge(getDataForLocal(), getDataForNet())
        .subscribe(new Observer<List<String>>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(List<String> strings) {
                for (String str: strings) { Log.i("JAVA", str); }
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
                Log.i("JAVA", "onComplete");
            }
        });

最後的列印結果是：

I/JAVA: 購物車的商品1

I/JAVA: 購物車的商品2

I/JAVA: 購物車的商品3

I/JAVA: onComplete

4.RxJava2 與 RxBinding 的使用

1.優化搜尋請求

當我們在 EditText 打字時搜尋的時候，可能使用者會打字很會快，那麼我們就沒有必要一直髮送網路請求，請求搜尋結果，我們可以通過當使用者打字停止後的延時500毫秒再發送搜尋請求：

// RxTextView.textChanges(edittext): Rxbinding用法
RxTextView.textChanges(editText)
        // 表示延時多少秒後執行，當你敲完字之後停下來的半秒就會執行下面語句
        .debounce(500, TimeUnit.MILLISECONDS)
        // 資料轉換 flatMap: 當同時多個數據請求訪問的時候，前面的網路資料會覆蓋後面的網路資料
        // 資料轉換 switchMap: 當同時多個網路請求訪問的時候，會以最後一個傳送請求為準，前面網路資料會被最後一個覆蓋
        .switchMap(new Function<CharSequence, ObservableSource<List<String>>>() {
            @Override
            public ObservableSource<List<String>> apply(
                    @NonNull CharSequence charSequence) throws Exception {
                // 網路請求操作，獲取我們需要的資料
                List<String> list = new ArrayList<String>();
                list.add("2017");
                list.add("2018");
                return Observable.just(list);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<List<String>>() {
            @Override
            public void accept(@NonNull List<String> strings) throws Exception {
                // 更新UI
                Log.i("JAVA", strings.toString());
            }
        });

2.優化點選請求

當用戶一直點選一個按鈕的時候，我們不應該一直呼叫訪問網路請求，而是 1秒內，只執行一次網路請求。

RxView.clicks(button).throttleFirst(1, TimeUnit.SECONDS)
        .subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {
            }
            @Override
            public void onNext(Object o) {
                Log.i("JAVA", "onClick");
            }
            @Override
            public void onError(Throwable e) {
            }
            @Override
            public void onComplete() {
            }
        });

5.RxJava2 踩過的一些坑

1.未解除訂閱而引起的記憶體洩漏

舉個例子，對於前面常用操作符 interval 做週期性操作的例子，並沒有使之停下來的，沒有去控制訂閱的生命週期，這樣，就有可能引發記憶體洩漏。所以，在 Activity 的 onDestroy() 方法執行的時候或者不需要繼續執行的時候應該解除訂閱。