Retrofit & RxJava 实战篇

一、网络请求实现：轮询、嵌套、合并数据、缓存、出错重试

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)//设置网络请求的Url地址
                .addConverterFactory(GsonConverterFactory.create())//设置数据解析器
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())//支持RxJava平台
                .client(client)
                .build();
        return retrofit;
    }

    //初始化请求数据参数
    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;
    }


    //无条件网络请求轮询
    private void interval() {
        //initialDelay：第一次延迟时间、period：间隔时间数字、unit:时间单位
        Observable.interval(2, 10, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                //网络请求具体实现
                getRequestInterface()
                        .registerLogin("login/login", initData())
                        .observeOn(AndroidSchedulers.mainThread())//回调在主线程
                        .subscribeOn(Schedulers.io())
                        .subscribe();//执行在io线程
            }
        }).subscribe();

        //start:事件序列起始点、count:事件数量、initialDelay:第1次事件延迟发送时间、period:间隔时间数字、unit:时间单位
        Observable.intervalRange(2, 10, 5, 10, TimeUnit.SECONDS).doOnNext(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) throws Exception {
                //网络请求具体实现code...
            }
        }).subscribe();
    }

    //有条件网络请求轮询
    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) {
                            // 此处选择发送onError事件以结束轮询
                            return Observable.error(new Throwable("轮询结束"));
                        }
                        //delay操作符延迟一段时间发送以实现轮询间间隔设置
                        return Observable.just(1).delay(5000, TimeUnit.MILLISECONDS);
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())//回调在主线程
                .subscribeOn(Schedulers.io())//执行在io线程
                .subscribe(new Consumer<RegisterLogin>() {
                    @Override
                    public void accept(RegisterLogin registerLogin) throws Throwable {
                        i[0]++;
                    }
                });
    }

    //网络请求嵌套回调
    private void doOnNext() {
        getRequestInterface()
                .registerLogin("login/login", initData())
                .observeOn(AndroidSchedulers.mainThread())//回调在主线程
                .subscribeOn(Schedulers.io())//执行在io线程
                .doOnNext(new Consumer<RegisterLogin>() {
                    @Override
                    public void accept(RegisterLogin registerLogin) throws Throwable {
                        //第1次网络请求成功
                    }
                })
                .observeOn(Schedulers.io())//新观察者切换到IO线程去发起请求
                .flatMap(new Function<RegisterLogin, ObservableSource<RegisterLogin>>() {
                    @Override
                    public ObservableSource<RegisterLogin> apply(RegisterLogin registerLogin) throws Throwable {
                        //registerLogin是网络请求1返回数据，发送网络请求2
                        return getRequestInterface()
                                .registerLogin("login/login", initData());
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())//切换到主线程处理网络请求2的结果
                .subscribe(new Consumer<RegisterLogin>() {
                    @Override
                    public void accept(RegisterLogin o) throws Throwable {
                        //o是网络请求2返回数据
                    }
                });
    }

    //采用 Merge()操作符合并数据
    private void merge() {
        Observable<String> network = Observable.just("网络");
        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()操作符合并数据（网络数据 + 网络数据）
    private void zip() {
        final Observable<RegisterLogin> observable1 = getRequestInterface()
                .registerLogin("login/login", initData())
                .subscribeOn(Schedulers.io());//执行在io线程
        final Observable<RegisterLogin> observable2 = getRequestInterface()
                .registerLogin("login/login", initData())
                .subscribeOn(Schedulers.io());//执行在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 {

                    }
                });
    }

    //从磁盘、内存缓存中获取缓存数据firstElement和concat
    private void firstElement() {
        final String memoryCache = null;
        final String diskCache = "从磁盘缓存中获取数据";
        Observable<String> memory = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                // 先判断内存缓存有无数据
                if (memoryCache != null) { // 若有该数据，则发送
                    emitter.onNext(memoryCache);
                } else { // 若无该数据，则直接发送结束事件
                    emitter.onComplete();
                }
            }
        });

        Observable<String> disk = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                // 先判断内存缓存有无数据
                if (diskCache != null) { // 若有该数据，则发送
                    emitter.onNext(diskCache);
                } else { // 若无该数据，则直接发送结束事件
                    emitter.onComplete();
                }
            }
        });

        Observable<String> observable = getRequestInterface()
                .getBody("login/login");//执行在io线程

        Observable.concat(memory, disk, observable)
                //通过firstElement()，从串联队列中取出并发送第1个有效事件（Next事件）
                //即依次判断检查memory、disk、network
                .firstElement()
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        //最终获取的数据来源
                    }
                });
    }

    // 可重试次数
    private int maxConnectCount = 10;
    // 当前已重试次数
    private int currentRetryCount = 0;
    // 重试等待时间
    private int waitRetryTime = 0;

    //网络请求出错重试
    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 {
                                //根据异常类型选择是否重试，属于IO异常，需重试
                                if (throwable instanceof IOException) {
                                    //当已重试次数 < 设置的重试次数才重试
                                    if (currentRetryCount < maxConnectCount) {
                                        // 记录重试次数
                                        currentRetryCount++;
                                        // 设置等待时间
                                        waitRetryTime = 1000 + currentRetryCount * 1000;
                                        return Observable.just(1).delay(waitRetryTime, TimeUnit.MILLISECONDS);
                                    } else {
                                        // 若重试次数已 > 设置重试次数，则不重试
                                        return Observable.error(new Throwable("重试次数已超过设置次数 = " + currentRetryCount + "，即 不再重试"));
                                    }
                                } else {
                                    // 若发生的异常不属于I/O异常，则不重试
                                    return Observable.error(new Throwable("发生了非网络异常（非I/O异常）"));
                                }
                            }
                        });
                    }
                })
                .subscribeOn(Schedulers.io())//执行在io线程
                .subscribe();
    }
}