Outline
[TOC]
前言
上個月RxJava2正式版發(fā)布了盲镶，但目前國內(nèi)的資料還比較少，以前使用過RxJava1只需要看看更新文檔就知道怎么使用了睁冬，但還有一些以前沒用過RxJava的朋友可能就不知道怎么辦了挎春，不知道該看RxJava1還是直接跳到RxJava2。所以寫下這個教程豆拨， 幫助那些沒有用過RxJava的朋友入門直奋。
注：如果你覺得寫得不好，請直接批評指出施禾。
我先回答這個問題：學(xué)習(xí)RxJava2需要先學(xué)習(xí)RxJava1嗎脚线？
這個問題就像論壇經(jīng)常問學(xué)Java前需要先學(xué)習(xí)C語言嗎，這里就不引戰(zhàn)了Ｃ指恪（PHP是世界上最好的語言Ｓ事獭！）
答案明顯不是攀例，如果你以前學(xué)過RxJava1船逮，那么對于RxJava2只需要看看更新了哪些東西就行了，其最核心的思想并沒有變化粤铭，如果你沒學(xué)過RxJava1傻唾，沒有關(guān)系，直接學(xué)習(xí)RxJava2。所以作為一個RxJava2的教程冠骄，本文中所有的名詞都屬于RxJava2中，并不涉及RxJava1加袋。
要在Android中使用RxJava2, 先添加Gradle配置:
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
正題
在開始學(xué)習(xí)之前凛辣，先來介紹點原理性的東西。
網(wǎng)上也有很多介紹RxJava原理的文章职烧，通常這些文章都從觀察者模式開始扁誓，先講觀察者，被觀察者蚀之，訂閱關(guān)系巴拉巴拉一大堆蝗敢，說實話，當(dāng)我第一次看到這些文章的時候已經(jīng)被這些名詞給繞暈了足删，用了很長的時間才理清楚它們之間的關(guān)系寿谴。可能是我太蠢了失受，境界不夠讶泰，領(lǐng)會不到那么多高大上的名詞.
今天我用兩根水管代替觀察者和被觀察者, 試圖用通俗易懂的話把它們的關(guān)系解釋清楚, 在這里我將從事件流這個角度來說明RxJava的基本工作原理。
先假設(shè)有兩根水管：
上面一根水管為事件產(chǎn)生的水管拂到，叫它上游吧痪署，下面一根水管為事件接收的水管叫它下游吧。
兩根水管通過一定的方式連接起來兄旬，使得上游每產(chǎn)生一個事件狼犯，下游就能收到該事件。注意這里和官網(wǎng)的事件圖是反過來的, 這里的事件發(fā)送的順序是先1,后2,后3這樣的順序, 事件接收的順序也是先1,后2,后3的順序, 我覺得這樣更符合我們普通人的思維, 簡單明了.
這里的上游和下游就分別對應(yīng)著RxJava中的Observable和Observer领铐，它們之間的連接就對應(yīng)著subscribe()悯森，因此這個關(guān)系用RxJava來表示就是：
//創(chuàng)建一個上游 Observable：
Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
});
//創(chuàng)建一個下游 Observer
Observer<Integer> observer = new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "complete");
}
};
//建立連接
observable.subscribe(observer);
這個運行的結(jié)果就是:
12-02 03:37:17.818 4166-4166/zlc.season.rxjava2demo D/TAG: subscribe
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 1
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 2
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: 3
12-02 03:37:17.819 4166-4166/zlc.season.rxjava2demo D/TAG: complete
注意: 只有當(dāng)上游和下游建立連接之后, 上游才會開始發(fā)送事件. 也就是調(diào)用了
subscribe()方法之后才開始發(fā)送事件.
把這段代碼連起來寫就成了RxJava引以為傲的鏈?zhǔn)讲僮鳎?/p>
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onNext(2);
emitter.onNext(3);
emitter.onComplete();
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "" + value);
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "complete");
}
});
接下來解釋一下其中兩個陌生的玩意：ObservableEmitter和Disposable.
ObservableEmitter： Emitter是發(fā)射器的意思，那就很好猜了罐孝，這個就是用來發(fā)出事件的呐馆，它可以發(fā)出三種類型的事件，通過調(diào)用emitter的onNext(T value)莲兢、onComplete()和onError(Throwable error)就可以分別發(fā)出next事件汹来、complete事件和error事件。
但是改艇，請注意收班，并不意味著你可以隨意亂七八糟發(fā)射事件，需要滿足一定的規(guī)則：
- 上游可以發(fā)送無限個onNext, 下游也可以接收無限個onNext.
- 當(dāng)上游發(fā)送了一個onComplete后, 上游onComplete之后的事件將會
繼續(xù)發(fā)送, 而下游收到onComplete事件之后將不再繼續(xù)接收事件. - 當(dāng)上游發(fā)送了一個onError后, 上游onError之后的事件將
繼續(xù)發(fā)送, 而下游收到onError事件之后將不再繼續(xù)接收事件. - 上游可以不發(fā)送onComplete或onError.
- 最為關(guān)鍵的是onComplete和onError必須唯一并且互斥, 即不能發(fā)多個onComplete, 也不能發(fā)多個onError, 也不能先發(fā)一個onComplete, 然后再發(fā)一個onError, 反之亦然
注: 關(guān)于onComplete和onError唯一并且互斥這一點, 是需要自行在代碼中進(jìn)行控制, 如果你的代碼邏輯中違背了這個規(guī)則, **并不一定會導(dǎo)致程序崩潰. ** 比如發(fā)送多個onComplete是可以正常運行的, 依然是收到第一個onComplete就不再接收了, 但若是發(fā)送多個onError, 則收到第二個onError事件會導(dǎo)致程序會崩潰.
以上幾個規(guī)則用示意圖表示如下:
| 示意圖 | |
|---|---|
| 只發(fā)送onNext事件 |  next
|
| 發(fā)送onComplete事件 |  complete
|
| 發(fā)送onError事件 |  error
|
介紹了ObservableEmitter, 接下來介紹Disposable, 這個單詞的字面意思是一次性用品,用完即可丟棄的. 那么在RxJava中怎么去理解它呢, 對應(yīng)于上面的水管的例子, 我們可以把它理解成兩根管道之間的一個機(jī)關(guān), 當(dāng)調(diào)用它的dispose()方法時, 它就會將兩根管道切斷, 從而導(dǎo)致下游收不到事件.
注意: 調(diào)用dispose()并不會導(dǎo)致上游不再繼續(xù)發(fā)送事件, 上游會繼續(xù)發(fā)送剩余的事件.
來看個例子, 我們讓上游依次發(fā)送1,2,3,complete,4,在下游收到第二個事件之后, 切斷水管, 看看運行結(jié)果:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit complete");
emitter.onComplete();
Log.d(TAG, "emit 4");
emitter.onNext(4);
}
}).subscribe(new Observer<Integer>() {
private Disposable mDisposable;
private int i;
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG, "subscribe");
mDisposable = d;
}
@Override
public void onNext(Integer value) {
Log.d(TAG, "onNext: " + value);
i++;
if (i == 2) {
Log.d(TAG, "dispose");
mDisposable.dispose();
Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
}
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onComplete() {
Log.d(TAG, "complete");
}
});
運行結(jié)果為:
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: subscribe
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 1
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: onNext: 2
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: dispose
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: isDisposed : true
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 3
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit complete
12-02 06:54:07.728 7404-7404/zlc.season.rxjava2demo D/TAG: emit 4
從運行結(jié)果我們看到, 在收到onNext 2這個事件后, 切斷了水管, 但是上游仍然發(fā)送了3, complete, 4這幾個事件, 而且上游并沒有因為發(fā)送了onComplete而停止. 同時可以看到下游的onSubscribe()方法是最先調(diào)用的.
Disposable的用處不止這些, 后面講解到了線程的調(diào)度之后, 我們會發(fā)現(xiàn)它的重要性. 隨著后續(xù)深入的講解, 我們會在更多的地方發(fā)現(xiàn)它的身影.
另外, subscribe()有多個重載的方法:
public final Disposable subscribe() {}
public final Disposable subscribe(Consumer<? super T> onNext) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}
public final void subscribe(Observer<? super T> observer) {}
最后一個帶有Observer參數(shù)的我們已經(jīng)使用過了,這里對其他幾個方法進(jìn)行說明.
- 不帶任何參數(shù)的
subscribe()表示下游不關(guān)心任何事件,你上游盡管發(fā)你的數(shù)據(jù)去吧, 老子可不管你發(fā)什么. - 帶有一個
Consumer參數(shù)的方法表示下游只關(guān)心onNext事件, 其他的事件我假裝沒看見, 因此我們?nèi)绻恍枰猳nNext事件可以這么寫:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
Log.d(TAG, "emit 1");
emitter.onNext(1);
Log.d(TAG, "emit 2");
emitter.onNext(2);
Log.d(TAG, "emit 3");
emitter.onNext(3);
Log.d(TAG, "emit complete");
emitter.onComplete();
Log.d(TAG, "emit 4");
emitter.onNext(4);
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.d(TAG, "onNext: " + integer);
}
});
- 其他幾個方法同理, 這里就不一一解釋了.
好了本次的教程到此結(jié)束, 明天周末, 祝大家周末愉快, 下周一再更新剩余的吧.
