歡迎轉載，轉載請注明出處：https://mp.csdn.net/mdeditor/80772129
落地98k,沒有倍鏡怪誰，讓你扶我你卻丢個手雷。
哈哈，大家好，喜歡裝逼的我又出現了，今天給大家帶來的是RxJava的相關讨論,RxJava已經出來很久了，也是一直在用，但是總感覺沒有完全掌握它，是以花了點時間也閱讀了很多文章以及官方的文檔，決定對其好好總結一番分享點有價值的文章，寫一個系列，對于RxJava1.x就不做讨論了過去的就讓它過去吧0.0，在寫本部落格時GitHub上最新的RxJava版本是2.1.16,位址:https://github.com/ReactiveX/RxJava，那麼本系列也就是從0-1帶着大家一起學習RxJava2.x的用法。如有疑問歡迎留言，如有謬誤歡迎批評指正。
通過本篇部落格你将學到以下知識點
①什麼是RxJava
②RxJava的優勢是什麼
③RxJava如何使用
④主要用來做什麼？
一、認識RxJava及RxJava的優勢
首先來看第一個問題，RxJava到底是個什麼東東？
官網給出的描述是這樣的：RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.
翻譯過來就是：RxJava是一個為使用可觀測的序列來組成異步的、基于事件的程式而産生的庫。說白了RxJava就是用來解決異步的，解決異步的方法有很多為什麼RxJava這麼流行呢？肯定它有它強于其它方法的優勢。那咱們來看看它的優勢是什麼？
1.函數式風格：對可觀察資料流使用無副作用的輸入輸出函數，避免了程式裡錯綜複雜的狀态
2.代碼書寫邏輯清晰：Rx的操作符通通常可以将複雜的難題簡化為邏輯非常簡單的代碼，可讀性非常強，随着程式邏輯的複雜，依然保持簡潔,解耦了各個子產品操作，單一化，不嵌套。
3.異步錯誤處理：傳統的try/catch沒辦法處理異步計算，Rx提供了合适的錯誤處理機制
4.輕松使用并發：Rx的Observables和Schedulers讓開發者可以擺脫底層的線程同步和各種并發問題
5.輕量級架構、支援Java 8 lambda、支援Java 6+和Android 2.3+
有沒有感覺很厲害，有木有被RxJava的氣場震懾到？
二、RxJava的使用
清楚了RxJava的定義以及優點之後，接着我們來看下RxJava如何使用，對于RxJava的使用首先必須明白以下三個概念
1.Observable(被觀察者)
2.Observer(觀察者)
3.Subscribe(訂閱)
為了更好的了解這三者之間的協作可以看下圖
其中上遊是Observable，下遊是Observer，上遊與下遊建立連接配接是通過Subscribe方法建立連接配接的，這樣了解是不是更好了解？
對于使用RxJava總共有三個步驟，可以參考上圖
①建立上遊的Observable
②建立下遊的Observer
③建立連接配接
說了這麼多，來搞一波代碼嘗嘗鹹淡
首先我們在Gradle中做如下配置
接着就可以肆無忌憚的裝逼了，按照上述說的三個步驟代碼如下
//建立上遊的Observable
Observable observable=Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
emitter.onNext("Hello");
emitter.onNext("Word");
emitter.onNext("!!!");
emitter.onComplete();
}
});
//建立下遊的Observer
Observer observer=new Observer<String>(){
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,"onSubscribe");
}
@Override
public void onNext(String s) {
Log.d(TAG,"onNext:"+s);
}
@Override
public void onError(Throwable e) {
Log.d(TAG,"onError");
}
@Override
public void onComplete() {
Log.d(TAG,"onComplete");
}
};
//連接配接上遊和下遊
observable.subscribe(observer);
列印日志如下：
我們來分析下執行的過程首先我們分别用代碼建立了上遊的Observable對象和下遊的Observer對象，建立好這兩個對象之後通過Observable對象的subscribe()方法将上遊與下遊建立了連接配接。建立連接配接成功之後會調用下遊observer的onSubscribe回調走onSubscribe方法預示着上遊與下遊建立了連接配接，此時列印了“onSubscribe”，之後上遊會依次發送三個事件，代碼如下
emitter.onNext("Hello");
emitter.onNext("Word");
emitter.onNext("!!!");
emitter.onComplete();
因為上遊與下遊已經建立了連接配接是以下遊會收到這三個事件，對應的會走下遊observer的onNext方法的回調，是以會依次列印”Hello”、”Word”、”!!!”,之後上遊發送發送emitter.onComplete();預示着上遊發送事件結束，此時會調用下遊的onComplete方法的回調，此時會列印onComplete。以上就是一個簡單的RxJava的小案例。
當然對于剛接觸到RxJava的同學可能還是有點陌生，那接下來咱們就來把這個案例了解透徹，首先來看看兩個對象ObservableEmitter和Disposable這兩個對象，其中Emitter是發射的意思，ObservableEmitter的源碼實作中共有三個方法:
onNext：用來發送資料，可多次調用，每調用一次發送一條資料
onError：用來發送異常通知，隻發送一次，若多次調用則第二次調用時報錯
onComplete：用來發送完成通知，隻發送一次，若多次調用隻發送第一條
在一個正确運作的事件序列中,onCompleted() 和 onError() 必須唯一并且互斥，資料在發送時，出現異常可以調用onError發送異常通知也可以不調用，因為其所在的方法subscribe會抛出異常，若資料在全部發送完之後均正常可以調用onComplete發送完成通知；其中，onError與onComplete不做強制性調用，并且兩者是事件序列中的最後一個。Observable可以發送無限個onNext, 觀察者也可以接收無限個onNext。Observable發送了一個onComplete（或者onError）後,可以繼續發送onComplete（或者onError）後續事件，但觀察者收到onComplete（或者onError）後不再接收事件。Observable可以不發送onComplete或onError。
對象Observer中的三個方法（onNext,onError,onComplete）正好與Emitter中的三個方法相對應，分别對Emitter中對應方法的行為作出響應。
Emitter調用onNext發送資料時，Observer會通過onNext接收資料。
Emitter調用onError發送異常通知時，Observer會通過onError接收異常通知，此時不再接收上遊發送的資料(此時上遊是可以發送資料的)
Emitter調用onComplete發送完成通知時，Observer會通過onComplete接收完成通知，并且不再接收上遊發送的資料(此時上遊是可以發送資料的)
1.中間發送一個oncomplete事件，原理圖如下：
代碼：
//建立上遊的Observable
Observable observable=Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG,"上遊 發射---->a");
emitter.onNext("發射---->a");
Log.d(TAG,"上遊 發射---->b");
emitter.onNext("發射---->b");
emitter.onComplete();
Log.d(TAG,"上遊 發射---->c");
emitter.onNext("發射---->c");
Log.d(TAG,"上遊 發射---->d");
emitter.onNext("發射---->d");
}
});
//建立下遊的Observer
Observer observer=new Observer<String>(){
@Override
public void onSubscribe(Disposable d) {
Log.d(TAG,"onSubscribe");
}
@Override
public void onNext(String s) {
Log.d(TAG,"下遊onNext接收:"+s);
}
@Override
public void onError(Throwable e) {
Log.d(TAG,"onError");
}
@Override
public void onComplete() {
Log.d(TAG,"onComplete");
}
};
//連接配接上遊和下遊
observable.subscribe(observer);
列印日志如下：
從日志中可以看出在上遊發送onComplete方法之後，上遊還是正常的發送事件，但是下遊卻沒有接收到上遊發送的事件。
2.中間發送一個onerror，原理圖如下：
代碼隻需修改上遊的發送事件的代碼：
//建立上遊的Observable
Observable observable=Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG,"上遊 發射---->a");
emitter.onNext("發射---->a");
Log.d(TAG,"上遊 發射---->b");
emitter.onNext("發射---->b");
emitter.onError(new NullPointerException());
Log.d(TAG,"上遊 發射---->c");
emitter.onNext("發射---->c");
Log.d(TAG,"上遊 發射---->d");
emitter.onNext("發射---->d");
}
});
日志：
從日志中可以看出在發送了onError事件後，上遊是可以發送資料的，但是下遊将不再接收上遊發送的資料。
3.發送多個onError或onComple
前面我們說到onComplete和onError必須唯一并且互斥若發送多個onComplete是可以正常運作的, 依然是收到第一個onComplete就不再接收了, 但若是發送多個onError, 則收到第二個onError事件會導緻程式會崩潰，這一點留給大家去驗證。
了解完ObservableEmitter之後我們再來了解一下Disposable 這個對象,Disposable這個對象是用來切斷上遊與下遊的連接配接的一個對象，切斷之後上遊可以繼續發送事件，但是下遊将不會再收到上遊發送的事件，廢話不多說咱們來看一段代碼：
//建立上遊的Observable
Observable observable=Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
Log.d(TAG,"上遊 發射---->a");
emitter.onNext("發射---->a");
Log.d(TAG,"上遊 發射---->b");
emitter.onNext("發射---->b");
Log.d(TAG,"上遊 發射---->c");
emitter.onNext("發射---->c");
Log.d(TAG,"上遊 發射---->d");
emitter.onNext("發射---->d");
}
});
//建立下遊的Observer
Observer observer=new Observer<String>(){
@Override
public void onSubscribe(Disposable d) {
disposable=d;
Log.d(TAG,"onSubscribe");
}
@Override
public void onNext(String s) {
if(s.equals("發射---->b")){
disposable.dispose();
}
Log.d(TAG,"下遊onNext接收:"+s);
}
@Override
public void onError(Throwable e) {
Log.d(TAG,"onError");
}
@Override
public void onComplete() {
Log.d(TAG,"onComplete");
}
};
//連接配接上遊和下遊
observable.subscribe(observer);
日志:
從日志中我們可以看出，在調用了dispose之後，上遊的發送是沒有受影響的，但是下遊收不到上遊發送的資料。
對于之前所述的代碼寫熟練之後就是經常說的鍊式調用
//RxJava鍊式調用
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
}
}).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
有木有感覺高大上？
四、真實項目中使用RxJava
到這裡可能部分同學對于RxJava可以用在什麼地方可能還不是特别清楚，其實RxJava的用處是非常非常大的，比如網絡請求、讀寫檔案等等，多處可以用到，因為本系列是從零開始學，是以這裡我就先舉個簡單的例子比如網絡請求RxJava可以與RxAndroid結合在一起使用對線程進行控制。代碼如下
//RxJava真實開發中的作用
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
//通過設定此方法的回調運作在子線程中，可以進行網絡請求等一些耗時的操作
//比如請求網絡拿到資料通過調用emitter.onNext(response);将請求的資料發送到下遊
}
}).subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<String>() {
//通過設定Observer運作在主線程，拿到網絡請求的資料進行解析使用
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
//在此接收上遊異步擷取的資料，比如網絡請求過來的資料進行處理
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
對于這個使用不懂沒關系先有個大緻的印象RxJava到底可以用在什麼地方，後續會有一系列關于RxJava的文章來講解具體的用法，記得關注和點贊，好了，相信通過本篇部落格大家會對RxJava有了一定的認識。鎖定本台敬請關注接下來的系列文章。
歡迎轉載，轉載請注明出處：https://mp.csdn.net/mdeditor/80772129
呵呵還沒完
參考文章：
https://mcxiaoke.gitbooks.io/rxdocs/content/Intro.html
https://www.jianshu.com/p/464fa025229e