RxJava2.0是一個非常棒的流式編程,采用的觀察者模式思想,事件的產(chǎn)生者產(chǎn)生事間之后發(fā)送給綁定的接受者,接受順序與發(fā)送順序一致.
<br />
依賴:
compile 'io.reactivex.rxjava2:rxjava:2.0.1'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'
<br />
簡單使用:
//觀察者模式,這里產(chǎn)生事件,事件產(chǎn)生后發(fā)送給接受者,但是一定要記得將事件的產(chǎn)生者和接收者捆綁在一起,否則會出現(xiàn)錯誤
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
//這里調(diào)用的方法會在產(chǎn)生事件之后會發(fā)送給接收者,接收者對應(yīng)方法會收到
e.onNext("hahaha");
e.onError(new Exception("wulala"));
e.onComplete();
}/*--
}).subscribe(new Observer<String>() {
//接受者,根據(jù)事件產(chǎn)生者產(chǎn)生的事件調(diào)用不同方法
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
}
@Override
public void onNext(String value) {
Log.e(TAG, "onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
我們來用圖解一下這其中發(fā)生了什么事:
上游朝下游發(fā)送數(shù)據(jù),經(jīng)過subscribe使上下游產(chǎn)生關(guān)系,即達(dá)成訂閱.
<br />
解析1:
ObservableEmitter,這是個啥東西?Emitter:顧名思義,即Rxjava的發(fā)射器,通過這個發(fā)射器,即可發(fā)送事件-----通過調(diào)用onNext,onError,onComplete方法發(fā)送不同事件.
注意:
雖然RxJava可以進(jìn)行事件發(fā)送,但這并不意味著你可以隨便發(fā)送,這其中需要遵循一些規(guī)則.
<br />
onNext:你可以發(fā)送無數(shù)個onNext,發(fā)送的每個onNext接受者都會接收到.
<br />
onError:當(dāng)發(fā)送了onError事件之后,發(fā)送者onError之后的事件依舊會繼續(xù)發(fā)送,但是接收者當(dāng)接收到onError之后就會停止接收事件了.
<br />
onComplete:當(dāng)發(fā)送了onComplete事件之后,發(fā)送者的onComplete之后的事件依舊會繼續(xù)發(fā)送,但是接收者接收到onComplete之后就停止接收事件了.
<br />
onError事件和onComplete事件是互斥的,但是這并不代表你配置了多個onError和onComplete一定會崩潰,多個onComplete是可以正常運(yùn)行的,但是只會接收到第一個,之后的就不會再接收到了,多個onError時,只會接收到第一個,第二個會直接造成程序崩潰.
<br />
解析2:
Disposable又是個啥東西,翻譯之后百度告訴我這東西叫做一次性的,是用來控制發(fā)送者和接受者之間的紐帶的,默認(rèn)為false,表示發(fā)送者和接受者直接的通信閥門關(guān)閉,可以正常通信,在調(diào)用dispose()方法之后,閥門開啟,會阻斷發(fā)送者和接收者之間的通信,從而斷開連接.
重載方法:
subscribe(); //表示發(fā)送者隨意發(fā)送數(shù)據(jù),接受者什么都不管,什么都不接收.
subscribe(Consumer<? super T> onNext) {} //只響應(yīng)onNext()事件,其他的事件忽略.
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} //含義同上
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {} //含義同上
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {} //含義同上
<br />
解析3:
默認(rèn)情況下,發(fā)送者和接收者都運(yùn)行在主線程,但是這顯然是不符合實際需求的,我們在日常使用中,通常用的最多的就是在子線程進(jìn)行各種耗時操作,然后發(fā)送到主線程進(jìn)行,難道我們就沒有辦法繼續(xù)用這個優(yōu)秀的庫了?想多了你,一個優(yōu)秀的庫如果連這都想不到,怎么能被稱為優(yōu)秀呢,RxJava中有線程調(diào)度器,通過線程調(diào)度器,我們可以很簡單的實現(xiàn)這種效果,下面放代碼.
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hahaha");
e.onNext("hahaha");
e.onNext("hahaha");
Log.e(TAG,"運(yùn)行在什么線程" + Thread.currentThread().getName());
e.onComplete();
}
}).subscribeOn(Schedulers.newThread()) //線程調(diào)度器,將發(fā)送者運(yùn)行在子線程
.observeOn(AndroidSchedulers.mainThread()) //接受者運(yùn)行在主線程
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
Log.e(TAG,"接收在什么線程" + Thread.currentThread().getName());
}
@Override
public void onNext(String value) {
Log.e(TAG, "onNext: " + value);
}
@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ", e);
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});
最終結(jié)果:
可以看到我們只加了兩行代碼,就實現(xiàn)了效果,還有比這個更優(yōu)秀,更簡單的么?
注意事項:
subscribeOn(),只有在第一次調(diào)用的時候生效,之后不管調(diào)用多少次,只會以第一次為準(zhǔn).
observeOn(),可以被調(diào)用多次,每次調(diào)用都會更改線程.
RxJava線程池中的幾個線程選項
- Schedulers.io() io操作的線程, 通常io操作,如文件讀寫.
- Schedulers.computation() 計算線程,適合高計算,數(shù)據(jù)量高的操作.
- Schedulers.newThread() 創(chuàng)建一個新線程,適合子線程操作.
- AndroidSchedulers.mainThread() Android的主線程,主線程
本文介紹簡單使用,下一篇開始講解RxJava中的操作符.
