Observable
Observable 是第一個核心類涩嚣。該類包含了 Rx 中的很多實(shí)現(xiàn)，以及所有核心的操作函數(shù)（operator妇垢、或者說 操作符）妓笙。
Observer/Consumer
Observer是一個接口，定義了onSubscribe(),onNext(),onError(),onComplete()可以操作Observable 發(fā)射的事件流
Subscriber 是 Observer 的一個實(shí)現(xiàn)幅狮。 Subscriber 實(shí)現(xiàn)了其他一些額外的功能募强，可以作為我們實(shí)現(xiàn) Observer 的基類。
Subject
Subject 是 Observable 的一個擴(kuò)展崇摄，同時還實(shí)現(xiàn)了 Observer 接口钻注。第一眼看上去可能有點(diǎn)奇怪，但是在有些場合下使用 Subject 將會非常便捷配猫。他們可以像 Observer 一樣接收事件幅恋，同時還可以像 Observable 一樣把接收到的事件再發(fā)射出去。這種特性非常適合 Rx 中的接入點(diǎn)泵肄，當(dāng)你的事件來至于 Rx 框架之外的代碼的時候捆交，你可以把這些數(shù)據(jù)先放到 Subject 中，然后再把 Subject轉(zhuǎn)換為一個 Observable腐巢，就可以在 Rx 中使用它們了品追。你可以把 Subject 當(dāng)做 Rx 中的 事件管道。
subscribe()
subscribe()是聯(lián)系Observable和Observer的橋梁冯丙。
使用RxJava的基本流程
1.創(chuàng)建Observable
Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
try {
e.onNext("A");
e.onNext("A");
e.onNext("A");
e.onComplete();
} catch (Exception ex) {
e.onError(ex);
}
}
});
2.創(chuàng)建Observer
Observer observer = new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
}
@Override
public void onNext(@NonNull String s) {
Log.i(TAG, "onNext: s = " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
};
3.subscribe()
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) {}
- 不帶任何參數(shù)的subscribe() 表示下游不關(guān)心任何事件
- 帶有一個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);
}
});
- 其余同理
這里我們使用最后一個
observable.subscribe(observer);
運(yùn)行結(jié)果:
08-24 08:03:25.041 6134-6134/com.example.wty.learnrxjava I/MainActivity: onSubscribe:
08-24 08:03:25.042 6134-6134/com.example.wty.learnrxjava I/MainActivity: onNext: s = A
08-24 08:03:25.042 6134-6134/com.example.wty.learnrxjava I/MainActivity: onNext: s = B
08-24 08:03:25.042 6134-6134/com.example.wty.learnrxjava I/MainActivity: onNext: s = C
08-24 08:03:25.042 6134-6134/com.example.wty.learnrxjava I/MainActivity: onComplete:
4.鏈?zhǔn)秸{(diào)用
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
try {
e.onNext("A");
e.onNext("B");
e.onNext("C");
e.onComplete();
} catch (Exception ex) {
e.onError(ex);
}
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
}
@Override
public void onNext(@NonNull String s) {
Log.i(TAG, "onNext: s = " + s);
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
});
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唯一并且互斥這一點(diǎn), 是需要自行在代碼中進(jìn)行控制, 如果你的代碼邏輯中違背了這個規(guī)則, 并不一定會導(dǎo)致程序崩潰. 比如發(fā)送多個onComplete是可以正常運(yùn)行的, 依然是收到第一個onComplete就不再接收了, 但若是發(fā)送多個onError, 則收到第二個onError事件會導(dǎo)致程序會崩潰。
介紹了ObservableEmitter, 接下來介紹Disposable, 這個單詞的字面意思是一次性用品,用完即可丟棄的. 那么在RxJava中怎么去理解它呢,對應(yīng)于觀察者模式相當(dāng)于取消訂閱, 對應(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,在下游收到第二個事件之后, 切斷水管, 看看運(yùn)行結(jié)果:
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onComplete();
e.onNext(4);
}
}).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(@NonNull Disposable d) {
Log.i(TAG, "onSubscribe: ");
disposable = d;
}
@Override
public void onNext(@NonNull Integer integer) {
Log.i(TAG, "onNext: integer = " + integer);
if (integer == 2) {
disposable.dispose();
}
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.i(TAG, "onComplete: ");
}
});
08-24 16:27:15.856 25856-25856/com.example.wty.learnrxjava I/MainActivity: onSubscribe:
08-24 16:27:15.857 25856-25856/com.example.wty.learnrxjava I/MainActivity: onNext: integer = 1
08-24 16:27:15.857 25856-25856/com.example.wty.learnrxjava I/MainActivity: onNext: integer = 2
