RxJava 2.x 教程及原始碼揭祕(一)入門理解及基本操作符
目錄
前言
本人從一年多前開始使用rxjava,可以說是見證了這個工具越來越流行,目前許多公司的android專案也都運用這個工具。Rxjava有著其獨特的優勢,也經常會被提及。對於這個東西,會用的人覺得他如此的好用,不理解的人卻覺得他非常難以上手。
那麼本文就將對Rxjava做一次詳細的探索,相信跟隨著博主的思路帶著自己的思考,你一定也能掌握使用這個熱門技術的要領。不僅如此,我們更要深入瞭解其內部原理,做到知其然,並且知其所以然。那麼,下面我們一起來揭開Rxjava神祕的面紗。
Rxjava的介紹
Rxjava在 GitHub 主頁上的自我介紹是 "a library for composing asynchronous and event-based programs using observable sequences for the Java VM"(一個在 Java VM 上使用可觀測的序列來組成非同步的、基於事件的程式的庫)。
從本質上看,這就是一個可以實現非同步操作的庫。
Rxjava的優勢
有的同學看完Rxjava的介紹不免產生了疑惑,非同步?那為什麼不用Handler,AsyncTask這些來實現呢?
Rxjava自然具備了其優勢,不然不可能會得到廣泛運用。
1.Rxjava符合響應式程式設計的思路,這讓他與另一款非常流行的網路請求框架Retrofit相輔相成,可以說android中Retrofit + OkHttp 是目前最主流的網路框架。
2.其另一個特點是支援函數語言程式設計,這保證了他的簡潔性。當你的業務需求越來越多,其他實現過程會越來越亂,產生迷之縮排。RxJava提供的一些方法,比如map,flatmap,可以很好的幫我們處理複雜的邏輯。Rxjava隨著程式邏輯變得越來越複雜,它依然能夠保持簡潔,這也是一個非常棒的優勢。
3.其實第三點嚴格來說也是第二點用一個優勢的不同體現,他對於任務的一些操作非常方便。比如非常方便我們執行緒的切換,無需手動建立子執行緒。
假設有這樣一個需求:介面上有一個自定義的檢視 imageCollectorView
,它的作用是顯示多張圖片,並能使用 addImage(Bitmap)
方法來任意增加顯示的圖片。現在需要程式將一個給出的目錄陣列 File[] folders
中每個目錄下的 png 圖片都加載出來並顯示在 imageCollectorView
中。需要注意的是,由於讀取圖片的這一過程較為耗時,需要放在後臺執行,而圖片的顯示則必須在 UI 執行緒執行。常用的實現方式有多種,我這裡貼出其中一種:
new Thread() {
@Override
public void run() {
super.run();
for (File folder : folders) {
File[] files = folder.listFiles();
for (File file : files) {
if (file.getName().endsWith(".png")) {
final Bitmap bitmap = getBitmapFromFile(file);
getActivity().runOnUiThread(new Runnable() {
@Override
public void run() {
imageCollectorView.addImage(bitmap);
}
});
}
}
}
}
}.start();
而如果使用 RxJava ,實現方式是這樣的:
Observable.from(folders)
.flatMap(new Func1<File, Observable<File>>() {
@Override
public Observable<File> call(File file) {
return Observable.from(file.listFiles());
}
})
.filter(new Func1<File, Boolean>() {
@Override
public Boolean call(File file) {
return file.getName().endsWith(".png");
}
})
.map(new Func1<File, Bitmap>() {
@Override
public Bitmap call(File file) {
return getBitmapFromFile(file);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Bitmap>() {
@Override
public void call(Bitmap bitmap) {
imageCollectorView.addImage(bitmap);
}
});
那位說話了:『你這程式碼明明變多了啊!簡潔個毛啊!』大兄弟你消消氣,我說的是邏輯的簡潔,不是單純的程式碼量少(邏輯簡潔才是提升讀寫程式碼速度的必殺技對不?)。觀察一下你會發現, RxJava 的這個實現,是一條從上到下的鏈式呼叫,沒有任何巢狀,這在邏輯的簡潔性上是具有優勢的。當需求變得複雜時,這種優勢將更加明顯(試想如果還要求只選取前 10 張圖片,常規方式要怎麼辦?如果有更多這樣那樣的要求呢?再試想,在這一大堆需求實現完兩個月之後需要改功能,當你翻回這裡看到自己當初寫下的那一片迷之縮排,你能保證自己將迅速看懂,而不是對著程式碼重新捋一遍思路?)。
另外,如果你的 IDE 是 Android Studio ,其實每次開啟某個 Java 檔案的時候,你會看到被自動 Lambda 化的預覽,這將讓你更加清晰地看到程式邏輯:
Observable.from(folders)
.flatMap((Func1) (folder) -> { Observable.from(file.listFiles()) })
.filter((Func1) (file) -> { file.getName().endsWith(".png") })
.map((Func1) (file) -> { getBitmapFromFile(file) })
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe((Action1) (bitmap) -> { imageCollectorView.addImage(bitmap) });
我們可以看到,通過Rxjava,所有程式碼都是鏈式展開的,一氣呵成,思路清晰,沒有了迷之縮
這裡就有人會擡槓,就為了看起來簡潔,我要費這麼大的勁學這東西。這裡我要多嘮叨幾句,作為一個有追求的程式設計師,我們應該有程式碼潔癖的好習慣,自己寫的程式碼,除了能滿足業務需求以外,還要追求程式碼的可讀性,思考如何讓其後期更好維護擴充套件,思考是否滿足六大基本原則,面向物件思想¥#%@*¥%@#。。。
說了一大堆,其實,在移動端,很多功能是大家都能實現的,然而真正區分大家水平的,是看你程式碼實現的過程是否合理,執行起來效能是否滿足要求。好的程式設計師寫的程式碼傻子都能看懂,而水平差的寫出來的程式碼只有自己能看懂。
一個剛學程式設計的人,你去跟他扯架構框架,設計模式,他怎麼可能理解,很多時候,程式碼多打幾行,這些自然就理解了。
Rxjava是觀察者模式
初學者可能會問,什麼是觀察者模式?
採用註冊(Register)或者稱為訂閱(Subscribe)的方式,告訴被觀察者:我需要你的某某狀態,你要在它變化的時候通知我。舉個安卓中最典型的例子吧,我們空間的點選事件監聽就是一種觀察者模式,對設定 OnClickListener 來說, View 是被觀察者, OnClickListener 是觀察者,二者通過 setOnClickListener() 方法達成訂閱關係。訂閱之後使用者點選按鈕的瞬間,Android Framework 就會將點選事件傳送給已經註冊的 OnClickListener 。
採取這樣被動的觀察方式,既省去了反覆檢索狀態的資源消耗,也能夠得到最高的反饋速度。
我們來對比下OnclickListener和RxJava
Buton持有OnClickListener的引用,當Button觸發了點選事件,呼叫OnClickListene的OnClik方法,這就是觀察者的流程,相信集合上面圖片的對比,你已經知道了RxJava中各部分分別對應的角色。
RxJava 有四個基本概念:Observable
(可觀察者,即被觀察者)、 Observer
(觀察者)、 subscribe
(訂閱)、事件。Observable
和 Observer
通過 subscribe()
方法實現訂閱關係,從而 Observable
可以在需要的時候發出事件來通知 Observer
。
除了OnNext()之外,還有其他事件。
onCompleted()
: 事件佇列完結。RxJava 不僅把每個事件單獨處理,還會把它們看做一個佇列。RxJava 規定,當不會再有新的onNext()
發出時,需要觸發onCompleted()
方法作為標誌。onError()
: 事件佇列異常。在事件處理過程中出異常時,onError()
會被觸發,同時佇列自動終止,不允許再有事件發出。- 在一個正確執行的事件序列中,
onCompleted()
和onError()
有且只有一個,並且是事件序列中的最後一個。需要注意的是,onCompleted()
和onError()
二者也是互斥的,即在佇列中呼叫了其中一個,就不應該再呼叫另一個
在 RxJava 2.x 中,Observable 用於訂閱 Observer,不再支援背壓(1.x 中可以使用背壓策略),而 Flowable 用於訂閱 Subscriber , 是支援背壓(Backpressure)的。
Rxjava是裝飾者模式
rxjava的所有操作,基本上都是lift實現的,運用的是裝飾者模式,這裡由於涉及到原理部分,不在此處展開,詳情看這裡
Observable
在 RxJava 1.x 中,我們最熟悉的莫過於 Observable
這個類了,在 RxJava 2.x 中建立了 一個 Observable
,Subscriber
都沒了,取而代之的是 ObservableEmmiter
,俗稱發射器。此外,由於沒有了Subscriber
的蹤影,我們建立觀察者時需使用 Observer
。而 Observer
也不是我們熟悉的那個 Observer
,又出現了一個 Disposable
引數。
RxJava 的三部曲
** 第一步:初始化 Observable **
** 第二步:初始化 Observer **
** 第三步:建立訂閱關係 **
Observable.create(new ObservableOnSubscribe<Integer>() { // 第一步:初始化Observable
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.e(TAG, "Observable emit 1" + "\n");
e.onNext(1);
Log.e(TAG, "Observable emit 2" + "\n");
e.onNext(2);
Log.e(TAG, "Observable emit 3" + "\n");
e.onNext(3);
e.onComplete();
Log.e(TAG, "Observable emit 4" + "\n" );
e.onNext(4);
}
}).subscribe(new Observer<Integer>() { // 第三步:訂閱
// 第二步:初始化Observer
private int i;
private Disposable mDisposable;
@Override
public void onSubscribe(@NonNull Disposable d) {
mDisposable = d;
}
@Override
public void onNext(@NonNull Integer integer) {
i++;
if (i == 2) {
// 在RxJava 2.x 中,新增的Disposable可以做到切斷的操作,讓Observer觀察者不再接收上游事件
mDisposable.dispose();
}
}
@Override
public void onError(@NonNull Throwable e) {
Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );
}
@Override
public void onComplete() {
Log.e(TAG, "onComplete" + "\n" );
}
});
不難看出,RxJava 2.x 與 1.x 還是存在著一些區別的。首先,建立 Observable
時,回撥的是 ObservableEmitter
,字面意思即發射器,並且直接 throws Exception。其次,在建立的 Observer 中,也多了一個回撥方法:onSubscribe
,傳遞引數為Disposable
,Disposable
相當於 RxJava 1.x 中的 Subscription
, 用於解除訂閱。可以看到示例程式碼中,在 i 自增到 2 的時候,訂閱關係被切斷。
07-03 14:24:11.663 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: onSubscribe : false
07-03 14:24:11.664 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: Observable emit 1
07-03 14:24:11.665 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: onNext : value : 1
07-03 14:24:11.666 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: Observable emit 2
07-03 14:24:11.667 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: onNext : value : 2
07-03 14:24:11.668 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: onNext : isDisposable : true
07-03 14:24:11.669 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: Observable emit 3
07-03 14:24:11.670 18467-18467/com.nanchen.rxjava2examples E/RxCreateActivity: Observable emit 4
RxJava的流程是這樣的
Rxjava的操作符
講了Rxjava的基本建立,那麼我們來了解下他的一些強大快捷的操作符,這也是為什麼Rxjava如此流行的一個原因
subScribeOn與observeOn切換執行緒
這兩者是我們執行緒切換的操作符,subScribeOn決定了被觀察者所在的執行緒也就是subscribe()
時所發生的執行緒,而observeOn決定觀察者所線上程也就是下游 Observer
回調發生的執行緒。。
執行緒切換需要注意的
RxJava 內建的執行緒排程器的確可以讓我們的執行緒切換得心應手,但其中也有些需要注意的地方。
- 簡單地說,
subscribeOn()
指定的就是發射事件的執行緒,observerOn
指定的就是訂閱者接收事件的執行緒。 - 多次指定發射事件的執行緒只有第一次指定的有效,也就是說多次呼叫
subscribeOn()
只有第一次的有效,其餘的會被忽略。 - 但多次指定訂閱者接收執行緒是可以的,也就是說每呼叫一次
observerOn()
,下游的執行緒就會切換一次。
Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
Log.e(TAG, "Observable thread is : " + Thread.currentThread().getName());
e.onNext(1);
e.onComplete();
}
}).subscribeOn(Schedulers.newThread())
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "After observeOn(mainThread),Current thread is " + Thread.currentThread().getName());
}
})
.observeOn(Schedulers.io())
.subscribe(new Consumer<Integer>() {
@Override
public void accept(@NonNull Integer integer) throws Exception {
Log.e(TAG, "After observeOn(io),Current thread is " + Thread.currentThread().getName());
}
});
輸出:
07-03 14:54:01.177 15121-15438/com.nanchen.rxjava2examples E/RxThreadActivity: Observable thread is : RxNewThreadScheduler-1
07-03 14:54:01.178 15121-15121/com.nanchen.rxjava2examples E/RxThreadActivity: After observeOn(mainThread),Current thread is main
07-03 14:54:01.179 15121-15439/com.nanchen.rxjava2examples E/RxThreadActivity: After observeOn(io),Current thread is RxCachedThreadScheduler-2
例項程式碼中,分別用 Schedulers.newThread()
和 Schedulers.io()
對發射執行緒進行切換,並採用 observeOn(AndroidSchedulers.mainThread()
和 Schedulers.io()
進行了接收執行緒的切換。可以看到輸出中發射執行緒僅僅響應了第一個 newThread
,但每呼叫一次 observeOn()
,執行緒便會切換一次,因此如果我們有類似的需求時,便知道如何處理了。
RxJava 中,已經內建了很多執行緒選項供我們選擇,例如有:
Schedulers.io()
代表io操作的執行緒, 通常用於網路,讀寫檔案等io密集型的操作;Schedulers.computation()
代表CPU計算密集型的操作, 例如需要大量計算的操作;Schedulers.newThread()
代表一個常規的新執行緒;AndroidSchedulers.mainThread()
代表Android的主執行緒
這些內建的 Scheduler
已經足夠滿足我們開發的需求,因此我們應該使用內建的這些選項,而 RxJava 內部使用的是執行緒池來維護這些執行緒,所以效率也比較高。
其他操作符
map實現關係轉換
map
操作符可以將一個 Observable
物件通過某種關係轉換為另一個Observable
物件。在 2.x 中和 1.x 中作用幾乎一致,不同點在於:2.x 將 1.x 中的 Func1
和 Func2
改為了 Function
和 BiFunction
。
採用 map 操作符進行網路資料解析
想必大家都知道,很多時候我們在使用 RxJava 的時候總是和 Retrofit 進行結合使用,而為了方便演示,這裡我們就暫且採用 OkHttp3 進行演示,配合 map
,doOnNext
,執行緒切換進行簡單的網路請求:
1)通過 Observable.create() 方法,呼叫 OkHttp 網路請求;
2)通過 map 操作符集合 gson,將 Response 轉換為 bean 類;
3)通過 doOnNext() 方法,解析 bean 中的資料,並進行資料庫儲存等操作;
4)排程執行緒,在子執行緒中進行耗時操作任務,在主執行緒中更新 UI ;
5)通過 subscribe(),根據請求成功或者失敗來更新 UI 。
Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Builder builder = new Builder()
.url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, MobileAddress>() {
@Override
public MobileAddress apply(@NonNull Response response) throws Exception {
if (response.isSuccessful()) {
ResponseBody body = response.body();
if (body != null) {
Log.e(TAG, "map:轉換前:" + response.body());
return new Gson().fromJson(body.string(), MobileAddress.class);
}
}
return null;
}
}).observeOn(AndroidSchedulers.mainThread())
.doOnNext(new Consumer<MobileAddress>() {
@Override
public void accept(@NonNull MobileAddress s) throws Exception {
Log.e(TAG, "doOnNext: 儲存成功:" + s.toString() + "\n");
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<MobileAddress>() {
@Override
public void accept(@NonNull MobileAddress data) throws Exception {
Log.e(TAG, "成功:" + data.toString() + "\n");
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "失敗:" + throwable.getMessage() + "\n");
}
});
flatMap()實現鏈式程式設計
flatMap()
: 這是一個很有用但非常難理解的變換,因此我決定花多些篇幅來介紹它。 首先假設這麼一種需求:假設有一個數據結構『學生』,現在需要打印出一組學生的名字。實現方式很簡單:
Student[] students = ...;
Subscriber<String> subscriber = new Subscriber<String>() {
@Override
public void onNext(String name) {
Log.d(tag, name);
}
...
};
Observable.from(students)
.map(new Func1<Student, String>() {
@Override
public String call(Student student) {
return student.getName();
}
})
.subscribe(subscriber);
很簡單。那麼再假設:如果要打印出每個學生所需要修的所有課程的名稱呢?(需求的區別在於,每個學生只有一個名字,但卻有多個課程。)首先可以這樣實現:
Student[] students = ...;
Subscriber<Student> subscriber = new Subscriber<Student>() {
@Override
public void onNext(Student student) {
List<Course> courses = student.getCourses();
for (int i = 0; i < courses.size(); i++) {
Course course = courses.get(i);
Log.d(tag, course.getName());
}
}
...
};
Observable.from(students)
.subscribe(subscriber);
依然很簡單。那麼如果我不想在 Subscriber
中使用 for 迴圈,而是希望 Subscriber
中直接傳入單個的 Course
物件呢(這對於程式碼複用很重要)?用 map()
顯然是不行的,因為 map()
是一對一的轉化,而我現在的要求是一對多的轉化。那怎麼才能把一個 Student 轉化成多個 Course 呢?
這個時候,就需要用 flatMap()
了:
Student[] students = ...;
Subscriber<Course> subscriber = new Subscriber<Course>() {
@Override
public void onNext(Course course) {
Log.d(tag, course.getName());
}
...
};
Observable.from(students)
.flatMap(new Func1<Student, Observable<Course>>() {
@Override
public Observable<Course> call(Student student) {
return Observable.from(student.getCourses());
}
})
.subscribe(subscriber);
從上面的程式碼可以看出, flatMap()
和 map()
有一個相同點:它也是把傳入的引數轉化之後返回另一個物件。但需要注意,和 map()
不同的是, flatMap()
中返回的是個 Observable
物件,並且這個 Observable
物件並不是被直接傳送到了 Subscriber
的回撥方法中。 flatMap()
的原理是這樣的:1. 使用傳入的事件物件建立一個 Observable
物件;2. 並不傳送這個 Observable
, 而是將它啟用,於是它開始傳送事件;3. 每一個創建出來的 Observable
傳送的事件,都被匯入同一個 Observable
,而這個 Observable
負責將這些事件統一交給 Subscriber
的回撥方法。這三個步驟,把事件拆成了兩級,通過一組新建立的 Observable
將初始的物件『鋪平』之後通過統一路徑分發了下去。而這個『鋪平』就是 flatMap()
所謂的 flat。
它可以把一個發射器 Observable
通過某種方法轉換為多個 Observables
,然後再把這些分散的 Observables
裝進一個單一的發射器 Observable
。但有個需要注意的是,flatMap
並不能保證事件的順序,如果需要保證,需要用到我們下面要講的 ConcatMap
。
flatMap()
示意圖:
flatMap 的實際應用
實現多個網路請求依次依賴
舉個例子。使用者註冊成功後需要自動登入,我們只需要先通過註冊介面註冊使用者資訊,註冊成功後馬上呼叫登入介面進行自動登入即可。
我們的 flatMap
恰好解決了這種應用場景,flatMap
操作符可以將一個發射資料的 Observable
變換為多個 Observables
,然後將它們發射的資料合併後放到一個單獨的 Observable
,利用這個特性,我們很輕鬆地達到了我們的需求。
Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows", 1 + "")
.build()
.getObjectObservable(FoodList.class) // 發起獲取食品列表的請求,並解析到FootList
.subscribeOn(Schedulers.io()) // 在io執行緒進行網路請求
.observeOn(AndroidSchedulers.mainThread()) // 在主執行緒處理獲取食品列表的請求結果
.doOnNext(new Consumer<FoodList>() {
@Override
public void accept(@NonNull FoodList foodList) throws Exception {
// 先根據獲取食品列表的響應結果做一些操作
Log.e(TAG, "accept: doOnNext :" + foodList.toString());
mRxOperatorsText.append("accept: doOnNext :" + foodList.toString()+"\n");
}
})
.observeOn(Schedulers.io()) // 回到 io 執行緒去處理獲取食品詳情的請求
.flatMap(new Function<FoodList, ObservableSource<FoodDetail>>() {
@Override
public ObservableSource<FoodDetail> apply(@NonNull FoodList foodList) throws Exception {
if (foodList != null && foodList.getTngou() != null && foodList.getTngou().size() > 0) {
return Rx2AndroidNetworking.post("http://www.tngou.net/api/food/show")
.addBodyParameter("id", foodList.getTngou().get(0).getId() + "")
.build()
.getObjectObservable(FoodDetail.class);
}
return null;
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodDetail>() {
@Override
public void accept(@NonNull FoodDetail foodDetail) throws Exception {
Log.e(TAG, "accept: success :" + foodDetail.toString());
mRxOperatorsText.append("accept: success :" + foodDetail.toString()+"\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "accept: error :" + throwable.getMessage());
mRxOperatorsText.append("accept: error :" + throwable.getMessage()+"\n");
}
});
zip 實現多個介面資料共同更新 UI
在實際應用中,我們極有可能會在一個頁面顯示的資料來源於多個介面,這時候我們的 zip
操作符為我們排憂解難。
zip
操作符可以將多個 Observable
的資料結合為一個數據源再發射出去。
Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512")
.build()
.getObjectObservable(MobileAddress.class);
Observable<CategoryResult> observable2 = Network.getGankApi()
.getCategoryData("Android",1,1);
Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() {
@Override
public String apply(@NonNull MobileAddress mobileAddress, @NonNull CategoryResult categoryResult) throws Exception {
return "合併後的資料為:手機歸屬地:"+mobileAddress.getResult().getMobilearea()+"人名:"+categoryResult.results.get(0).who;
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String s) throws Exception {
Log.e(TAG, "accept: 成功:" + s+"\n");
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "accept: 失敗:" + throwable+"\n");
}
});
concat決定邏輯順序
concat
可以做到不交錯的發射兩個甚至多個 Observable 的發射事件,並且只有前一個 Observable
終止(onComplete
) 後才會訂閱下一個 Observable
。
採用 concat 操作符先讀取快取再通過網路請求獲取資料
想必在實際應用中,很多時候(對資料操作不敏感時)都需要我們先讀取快取的資料,如果快取沒有資料,再通過網路請求獲取,隨後在主執行緒更新我們的UI。
concat
操作符簡直就是為我們這種需求量身定做。
利用 concat
的必須呼叫 onComplete
後才能訂閱下一個 Observable
的特性,我們就可以先讀取快取資料,倘若獲取到的快取資料不是我們想要的,再呼叫 onComplete()
以執行獲取網路資料的 Observable
,如果快取資料能應我們所需,則直接呼叫 onNext()
,防止過度的網路請求,浪費使用者的流量。
Observable<FoodList> cache = Observable.create(new ObservableOnSubscribe<FoodList>() {
@Override
public void subscribe(@NonNull ObservableEmitter<FoodList> e) throws Exception {
Log.e(TAG, "create當前執行緒:"+Thread.currentThread().getName() );
FoodList data = CacheManager.getInstance().getFoodListData();
// 在操作符 concat 中,只有呼叫 onComplete 之後才會執行下一個 Observable
if (data != null){ // 如果快取資料不為空,則直接讀取快取資料,而不讀取網路資料
isFromNet = false;
Log.e(TAG, "\nsubscribe: 讀取快取資料:" );
runOnUiThread(new Runnable() {
@Override
public void run() {
mRxOperatorsText.append("\nsubscribe: 讀取快取資料:\n");
}
});
e.onNext(data);
}else {
isFromNet = true;
runOnUiThread(new Runnable() {
@Override
public void run() {
mRxOperatorsText.append("\nsubscribe: 讀取網路資料:\n");
}
});
Log.e(TAG, "\nsubscribe: 讀取網路資料:" );
e.onComplete();
}
}
});
Observable<FoodList> network = Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list")
.addQueryParameter("rows",10+"")
.build()
.getObjectObservable(FoodList.class);
// 兩個 Observable 的泛型應當保持一致
Observable.concat(cache,network)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<FoodList>() {
@Override
public void accept(@NonNull FoodList tngouBeen) throws Exception {
Log.e(TAG, "subscribe 成功:"+Thread.currentThread().getName() );
if (isFromNet){
mRxOperatorsText.append("accept : 網路獲取資料設定快取: \n");
Log.e(TAG, "accept : 網路獲取資料設定快取: \n"+tngouBeen.toString() );
CacheManager.getInstance().setFoodListData(tngouBeen);
}
mRxOperatorsText.append("accept: 讀取資料成功:" + tngouBeen.toString()+"\n");
Log.e(TAG, "accept: 讀取資料成功:" + tngouBeen.toString());
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
Log.e(TAG, "subscribe 失敗:"+Thread.currentThread().getName() );
Log.e(TAG, "accept: 讀取資料失敗:"+throwable.getMessage() );
mRxOperatorsText.append("accept: 讀取資料失敗:"+throwable.getMessage()+"\n");
}
});
有時候我們的快取可能還會分為 memory 和 disk ,實際上都差不多,無非是多寫點 Observable
,然後通過 concat
合併即可。
採用 interval 操作符實現心跳間隔任務
這裡就簡單的意思一下輪訓。
private Disposable mDisposable;
@Override
protected void doSomething() {
mDisposable = Flowable.interval(1, TimeUnit.SECONDS)
.doOnNext(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
Log.e(TAG, "accept: doOnNext : "+aLong );
}
})
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Long>() {
@Override
public void accept(@NonNull Long aLong) throws Exception {
Log.e(TAG, "accept: 設定文字 :"+aLong );
mRxOperatorsText.append("accept: 設定文字 :"+aLong +"\n");
}
});
}
/**
* 銷燬時停止心跳
*/
@Override
protected void onDestroy() {
super.onDestroy();
if (mDisposable != null){
mDisposable.dispose();
}
}
補充
到這裡,你就瞭解了Rxjava的基本特性以及它的一些操作技巧,相信認真看完,你也就能感受到Rxjava的強大之處。
而且,Rxjava配合Retrofit來使用,更能發揮它響應式程式設計的特點,此外,RxBinding,RxBus也是一些不錯的擴充套件,這裡由於篇幅有限不做展開。
當然,作為一個有抱負的程式設計師,我們不應僅僅滿足於會用,還要去了解他的內部實現原理,從中看出其包涵的一些程式設計思想,去學習他們對設計模式的一些運用。這裡由於篇幅太長了,關於Rxjava的原理部分我會另寫一篇,帶你透徹理解Rxjava是如何實現執行緒的切換,以及如何通過map,flatmap這些操作符來包裝我們的業務邏輯。
本文部分借鑑 https://www.jianshu.com/p/0cd258eecf60