RxJava——目前最熱門(mén)的響應(yīng)式函數(shù)編程框架拱燃。
本文主要總結(jié)了筆者在項(xiàng)目中使用到的RxJava的場(chǎng)景，部分例子參考自網(wǎng)絡(luò)
[筆者仍為Android初學(xué)者弄跌。如有解釋錯(cuò)誤的地方糠溜，歡迎評(píng)論區(qū)指正探討]
本文主要介紹一些RxJava的使用場(chǎng)景，為了完整介紹侮东，本文不使用Retrofit或者將RxJava進(jìn)行簡(jiǎn)單的封裝，原汁原味铁蹈。
當(dāng)然，如果結(jié)合Retrofit览徒，下述的代碼會(huì)簡(jiǎn)潔很多措嵌。
單個(gè)網(wǎng)絡(luò)請(qǐng)求數(shù)據(jù)并更新UI
這個(gè)比較簡(jiǎn)單枫慷，整個(gè)流程大致是：
- 通過(guò)
Obsrvable.create方法罗丰，調(diào)用OkHttp網(wǎng)絡(luò)請(qǐng)求 - 通過(guò)
map方法結(jié)合gson，將response轉(zhuǎn)換為bean類(lèi) - 通過(guò)
onNext，解析bean中數(shù)據(jù)，并進(jìn)行數(shù)據(jù)庫(kù)存儲(chǔ) - 調(diào)度線程
- 通過(guò)
subscribe萝快，根據(jù)請(qǐng)求成功或異常來(lái)更新UI
Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("url")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, Bean>() {
@Override
public Bean apply(@NonNull Response response) throws Exception {
//Gson
}
}).doOnNext(new Consumer<Bean>() {
@Override
public void accept(@NonNull Bean bean) throws Exception {
//saveData
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Bean>() {
@Override
public void accept(@NonNull Bean bean) throws Exception {
//refresh UI
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
//get ERROR
}
});
多個(gè)網(wǎng)絡(luò)請(qǐng)求依次依賴
這里主要是依賴于flatMap關(guān)鍵字，FlatMap可以將一個(gè)發(fā)射數(shù)據(jù)的Observable變換為多個(gè)Observables，然后將它們發(fā)射的數(shù)據(jù)合并后放進(jìn)一個(gè)單獨(dú)的Observable。
利用這個(gè)特性奕谭，我們可以將Observable轉(zhuǎn)成另一個(gè)Observable
Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("url")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
}).map(new Function<Response, FirstBean>() {
@Override
public FirstBean apply(@NonNull Response response) throws Exception {
//Gson
}
}).flatMap(new Function<FirstBean, ObservableSource<Response>>() {
@Override
public ObservableSource<Response> apply(@NonNull FirstBean bean) throws Exception {
final String s = bean.getData();
return Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("url/" + s)
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
}
});
}
}).map(new Function<Response, SecondBean>() {
@Override
public SecondBean apply(@NonNull Response response) throws Exception {
//Gson
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<SecondBean>() {
@Override
public void accept(@NonNull SecondBean secondBean) throws Exception {
//refresh UI
}
});
先讀取緩存數(shù)據(jù)并展示UI再獲取網(wǎng)絡(luò)數(shù)據(jù)刷新UI
這里需要依賴另一個(gè)操作符：Concat
concat可以做到不交錯(cuò)的發(fā)射兩個(gè)或多個(gè)Observable的發(fā)射物根吁，并且只有前一個(gè)Observable終止(onComleted)才會(huì)訂閱下一個(gè)Obervable
利用這個(gè)特性沃斤，我們就可以依次的讀取緩存數(shù)據(jù)展示UI，然后再獲取網(wǎng)絡(luò)數(shù)據(jù)刷新UI
- 首先創(chuàng)建一個(gè)從cache獲取數(shù)據(jù)的observable
- 再創(chuàng)建一個(gè)從網(wǎng)絡(luò)獲取數(shù)據(jù)的Observable(可以通過(guò)map等方法轉(zhuǎn)換數(shù)據(jù)類(lèi)型)
- 通過(guò)concat方法將多個(gè)observable結(jié)合起來(lái)
- 通過(guò)subscribe訂閱每一個(gè)observable
Observable<List<String>> cache = Observable.create(new ObservableOnSubscribe<List<String>>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
CacheManager manager = CacheManager.getInstance();
List<String> data = manager.query();
e.onNext(data);
//一定要有onComplete十厢，不然不會(huì)執(zhí)行第二個(gè)Observale
e.onComplete();
}
});
Observable<List<String>> network = Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("url")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
e.onComplete();
}
}).map(new Function<Response, List<String>>() {
@Override
public List<String> apply(@NonNull Response response) throws Exception {
//解析數(shù)據(jù)
}
});
//兩個(gè)observable的泛型應(yīng)該保持一致
Observable.concat(cache, network)
.subscribeOn(Schedulers.io())
.subscribe(new Consumer<List<String>>() {
@Override
public void accept(@NonNull List<String> strings) throws Exception {
//refresh ui
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
//get error
}
});
獲取網(wǎng)絡(luò)數(shù)據(jù)前先讀取緩存
其實(shí)和上面的那種類(lèi)似，只需要稍微修改一下邏輯即可：
當(dāng)緩存的Observable獲取到數(shù)據(jù)時(shí)，只執(zhí)行onNext，獲取不到則只執(zhí)行onComplete
Observable<String> cache = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
CacheManager manager = CacheManager.getInstance();
String data = manager.queryForPosition(0);
if (data != null) {
e.onNext(data);
} else {
//調(diào)用onComplete之后會(huì)執(zhí)行下一個(gè)Observable
//如果緩存為空，那么直接結(jié)束，進(jìn)行網(wǎng)絡(luò)請(qǐng)求
e.onComplete();
}
}
});
Observable<String> network = Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("url")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
e.onComplete();
}
}).map(new Function<Response, String>() {
@Override
public String apply(@NonNull Response response) throws Exception {
//解析數(shù)據(jù)
}
});
//兩個(gè)observable的泛型應(yīng)該保持一致
Observable.concat(cache, network)
.subscribeOn(Schedulers.io())
.subscribe(new Consumer<String>() {
@Override
public void accept(@NonNull String strings) throws Exception {
//refresh ui
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
//get error
}
});
當(dāng)然萄涯，有的時(shí)候我們的緩存可能還會(huì)分為memory和disk争占，無(wú)差，只需要多寫(xiě)一個(gè)Observable然后一樣通過(guò)concat合并即可。
結(jié)合多個(gè)接口的數(shù)據(jù)再更新UI
這個(gè)時(shí)候就需要靠zip方法啦红伦，zip方法可以將多個(gè)Observable的數(shù)據(jù)結(jié)合為一個(gè)數(shù)據(jù)源再發(fā)射出去膨桥。
Observable<FirstBean> firstRequest = Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("firstUrl")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
e.onComplete();
}
}).map(new Function<Response, FirstBean>() {
@Override
public FirstBean apply(@NonNull Response response) throws Exception {
//解析數(shù)據(jù)
}
});
Observable<SecondBean> secondRequest = Observable.create(new ObservableOnSubscribe<Response>() {
@Override
public void subscribe(@NonNull ObservableEmitter e) throws Exception {
Request.Builder builder = new Request.Builder()
.url("secondUrl")
.get();
Request request = builder.build();
Call call = new OkHttpClient().newCall(request);
Response response = call.execute();
e.onNext(response);
e.onComplete();
}
}).map(new Function<Response, SecondBean>() {
@Override
public SecondBean apply(@NonNull Response response) throws Exception {
//解析數(shù)據(jù)
}
});
Observable.zip(firstRequest, secondRequest, new BiFunction<FirstBean, SecondBean, WholeBean>() {
@Override
public WholeBean apply(@NonNull FirstBean firstBean, @NonNull SecondBean secondBean) throws Exception {
//結(jié)合數(shù)據(jù)為一體
}
})
.subscribeOn(Schedulers.io())
.subscribe(new Consumer<WholeBean>() {
@Override
public void accept(@NonNull WholeBean strings) throws Exception {
//refresh ui
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
//get error
}
});
當(dāng)然，如果你的兩個(gè)api返回的是相同類(lèi)型的數(shù)據(jù)，那么可以直接使用merge將數(shù)據(jù)合并，而不需要實(shí)現(xiàn)回調(diào)募狂。
減少頻繁的網(wǎng)絡(luò)請(qǐng)求
設(shè)想一種場(chǎng)景：點(diǎn)擊一次button就進(jìn)行一次網(wǎng)絡(luò)請(qǐng)求勺三，或者當(dāng)輸入框數(shù)據(jù)變化時(shí)進(jìn)行網(wǎng)絡(luò)請(qǐng)求，那么這樣就會(huì)在一下子產(chǎn)生大量的網(wǎng)絡(luò)請(qǐng)求，但實(shí)際上又沒(méi)有必要扫沼，這個(gè)時(shí)候就可以通過(guò)debounce方法來(lái)處理器罐，debounce操作符會(huì)過(guò)濾掉發(fā)射速率過(guò)快的數(shù)據(jù)項(xiàng)：
為了方便處理點(diǎn)擊事件和Observable的關(guān)系，我們引入RxBinding處理：
RxView.clicks(mButton)
.debounce(2, TimeUnit.SECONDS)
.subscribe(new Consumer<Object>() {
@Override
public void accept(@NonNull Object o) throws Exception {
// refresh ui
}
}, new Consumer<Throwable>() {
@Override
public void accept(@NonNull Throwable throwable) throws Exception {
// get error
}
});
