一、網(wǎng)絡請求實現(xiàn)：輪詢、嵌套坯癣、合并數(shù)據(jù)哼勇、緩存都伪、出錯重試
public class RetrofitRxJavaStudy {
private static NetworkInterface networkInterface;
//獲取接口類NetworkInterface并初始化Retrofit
private static NetworkInterface getRequestInterface() {
if (networkInterface == null) {
synchronized (NetworkInterface.class) {
networkInterface = initRetrofit().create(NetworkInterface.class);
}
}
return networkInterface;
}
private static Retrofit initRetrofit() {
OkHttpClient client = new OkHttpClient.Builder()
.build();
Retrofit retrofit = new Retrofit.Builder()
.baseUrl(NetworkInterface.HOST)//設置網(wǎng)絡請求的Url地址
.addConverterFactory(GsonConverterFactory.create())//設置數(shù)據(jù)解析器
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())//支持RxJava平臺
.client(client)
.build();
return retrofit;
}
//初始化請求數(shù)據(jù)參數(shù)
private RequestBody initData() {
Gson gson = new Gson();
Map<String, String> map = new HashMap<String, String>();
map.put("phone", "phone");
map.put("code", "code");
String strJson = gson.toJson(map);
RequestBody body = RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"), strJson);
return body;
}
//無條件網(wǎng)絡請求輪詢
private void interval() {
//initialDelay：第一次延遲時間、period：間隔時間數(shù)字积担、unit:時間單位
Observable.interval(2, 10, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
//網(wǎng)絡請求具體實現(xiàn)
getRequestInterface()
.registerLogin("login/login", initData())
.observeOn(AndroidSchedulers.mainThread())//回調在主線程
.subscribeOn(Schedulers.io())
.subscribe();//執(zhí)行在io線程
}
}).subscribe();
//start:事件序列起始點陨晶、count:事件數(shù)量、initialDelay:第1次事件延遲發(fā)送時間帝璧、period:間隔時間數(shù)字先誉、unit:時間單位
Observable.intervalRange(2, 10, 5, 10, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
@Override
public void accept(Long aLong) throws Exception {
//網(wǎng)絡請求具體實現(xiàn)code...
}
}).subscribe();
}
//有條件網(wǎng)絡請求輪詢
private void repeatWhen() {
final int[] i = {0};
getRequestInterface()
.registerLogin("login/login", initData())
.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(Observable<Object> objectObservable) throws Throwable {
if (i[0] > 3) {
// 此處選擇發(fā)送onError事件以結束輪詢
return Observable.error(new Throwable("輪詢結束"));
}
//delay操作符延遲一段時間發(fā)送以實現(xiàn)輪詢間間隔設置
return Observable.just(1).delay(5000, TimeUnit.MILLISECONDS);
}
})
.observeOn(AndroidSchedulers.mainThread())//回調在主線程
.subscribeOn(Schedulers.io())//執(zhí)行在io線程
.subscribe(new Consumer<RegisterLogin>() {
@Override
public void accept(RegisterLogin registerLogin) throws Throwable {
i[0]++;
}
});
}
//網(wǎng)絡請求嵌套回調
private void doOnNext() {
getRequestInterface()
.registerLogin("login/login", initData())
.observeOn(AndroidSchedulers.mainThread())//回調在主線程
.subscribeOn(Schedulers.io())//執(zhí)行在io線程
.doOnNext(new Consumer<RegisterLogin>() {
@Override
public void accept(RegisterLogin registerLogin) throws Throwable {
//第1次網(wǎng)絡請求成功
}
})
.observeOn(Schedulers.io())//新觀察者切換到IO線程去發(fā)起請求
.flatMap(new Function<RegisterLogin, ObservableSource<RegisterLogin>>() {
@Override
public ObservableSource<RegisterLogin> apply(RegisterLogin registerLogin) throws Throwable {
//registerLogin是網(wǎng)絡請求1返回數(shù)據(jù)，發(fā)送網(wǎng)絡請求2
return getRequestInterface()
.registerLogin("login/login", initData());
}
})
.observeOn(AndroidSchedulers.mainThread())//切換到主線程處理網(wǎng)絡請求2的結果
.subscribe(new Consumer<RegisterLogin>() {
@Override
public void accept(RegisterLogin o) throws Throwable {
//o是網(wǎng)絡請求2返回數(shù)據(jù)
}
});
}
//采用 Merge()操作符合并數(shù)據(jù)
private void merge() {
Observable<String> network = Observable.just("網(wǎng)絡");
Observable<String> file = Observable.just("本地文件");
Observable.merge(network, file).subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String value) {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
}
});
}
//采用Zip()操作符合并數(shù)據(jù)（網(wǎng)絡數(shù)據(jù) + 網(wǎng)絡數(shù)據(jù)）
private void zip() {
final Observable<RegisterLogin> observable1 = getRequestInterface()
.registerLogin("login/login", initData())
.subscribeOn(Schedulers.io());//執(zhí)行在io線程
final Observable<RegisterLogin> observable2 = getRequestInterface()
.registerLogin("login/login", initData())
.subscribeOn(Schedulers.io());//執(zhí)行在io線程
Observable.zip(observable1, observable2, new BiFunction<RegisterLogin, RegisterLogin, Object>() {
@Override
public Object apply(RegisterLogin registerLogin, RegisterLogin registerLogin2) throws Throwable {
return registerLogin.getMessage() + " & " + registerLogin2.getMessage();
}
}).observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<Object>() {
@Override
public void accept(Object o) throws Throwable {
}
});
}
//從磁盤的烁、內存緩存中獲取緩存數(shù)據(jù)firstElement和concat
private void firstElement() {
final String memoryCache = null;
final String diskCache = "從磁盤緩存中獲取數(shù)據(jù)";
Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
// 先判斷內存緩存有無數(shù)據(jù)
if (memoryCache != null) { // 若有該數(shù)據(jù)褐耳，則發(fā)送
emitter.onNext(memoryCache);
} else { // 若無該數(shù)據(jù)，則直接發(fā)送結束事件
emitter.onComplete();
}
}
});
Observable<String> disk = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> emitter) throws Exception {
// 先判斷內存緩存有無數(shù)據(jù)
if (diskCache != null) { // 若有該數(shù)據(jù)渴庆，則發(fā)送
emitter.onNext(diskCache);
} else { // 若無該數(shù)據(jù)铃芦，則直接發(fā)送結束事件
emitter.onComplete();
}
}
});
Observable<String> observable = getRequestInterface()
.getBody("login/login");//執(zhí)行在io線程
Observable.concat(memory, disk, observable)
//通過firstElement()雅镊，從串聯(lián)隊列中取出并發(fā)送第1個有效事件（Next事件）
//即依次判斷檢查memory、disk刃滓、network
.firstElement()
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
//最終獲取的數(shù)據(jù)來源
}
});
}
// 可重試次數(shù)
private int maxConnectCount = 10;
// 當前已重試次數(shù)
private int currentRetryCount = 0;
// 重試等待時間
private int waitRetryTime = 0;
//網(wǎng)絡請求出錯重試
private void flatMap() {
getRequestInterface()
.registerLogin("login/login", initData())
.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {
return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
@Override
public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {
//根據(jù)異常類型選擇是否重試仁烹，屬于IO異常，需重試
if (throwable instanceof IOException) {
//當已重試次數(shù) < 設置的重試次數(shù)才重試
if (currentRetryCount < maxConnectCount) {
// 記錄重試次數(shù)
currentRetryCount++;
// 設置等待時間
waitRetryTime = 1000 + currentRetryCount * 1000;
return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);
} else {
// 若重試次數(shù)已 > 設置重試次數(shù)注盈，則不重試
return Observable.error(new Throwable("重試次數(shù)已超過設置次數(shù) = " + currentRetryCount + "晃危，即 不再重試"));
}
} else {
// 若發(fā)生的異常不屬于I/O異常，則不重試
return Observable.error(new Throwable("發(fā)生了非網(wǎng)絡異常（非I/O異常）"));
}
}
});
}
})
.subscribeOn(Schedulers.io())//執(zhí)行在io線程
.subscribe();
}
}
最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯(lián)系作者
