RxJava2 实战知识梳理(11) 检测网络状态并自动重试请求

1、应用背景

今天，咱们以一个请求天气数据的例子，来演示如何用RxJava实现网络重连时的自动请求，首先，咱们对这个需求进行一个简单的描述，整个项目的框架以下所示：

• 在应用启动时，咱们会启动定位模块，该定位模块在后台每隔一段时间发起一次定位请求，拿到定位的结果后，咱们经过该城市向服务器发起请求，以获取对应城市的天气信息进行展现。
• 可是在拿到城市以后向服务器请求天气的过程当中有多是处于没有网络的状态，致使没法获取城市的天气信息并刷新界面，所以，咱们须要检测网络的状态，在网络重连的时候，读取上一次缓存的城市，向服务器发起请求以获取城市对应天气信息。

本文的示例代码在 RxSample 的第十一章中。

2、示例

2.1 定位模块

咱们经过一个后台线程来模拟定位的过程，它每隔一段时间获取一次定位的结果，并将该结果经过mCityPublish发送数据给它的订阅者。

//用于发布定位到的城市结果。
    private PublishSubject<Long> mCityPublish;

    //模拟定位模块的回调。
    private void startUpdateLocation() {
        mLocationThread = new Thread() {
            @Override
            public void run() {
                while (true) {
                    try {
                        for (long cityId : CITY_ARRAY) {
                            if (isInterrupted()) {
                                break;
                            }
                            Log.d(TAG, "从新定位");
                            Thread.sleep(5000);
                            Log.d(TAG, "定位到城市信息=" + cityId);
                            mCityPublish.onNext(cityId);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        mLocationThread.start();
    }
复制代码

在mCityPublish发送消息到订阅者收到消息之间，咱们还须要作一些特殊的处理：

private Observable<Long> getCityPublish() {
        return mCityPublish.distinctUntilChanged().doOnNext(new Consumer<Long>() {

            @Override
            public void accept(Long aLong) throws Exception {
                saveCacheCity(aLong);
            }

        });
    }
复制代码

这里咱们作了两步处理：

• 使用distinctUntilChanged对定位结果进行过滤，若是这次定位的结果和上次定位的结果相同，那么不通知订阅者。distinctUntilChanged的原理图以下所示：

  

• 使用doOnNext，在返回结果给订阅者以前，先把最新一次的定位结果存储起来，用于在以后网络重连以后进行请求。

2.2 网络状态模块

与定位模块相似，咱们也须要一个mNetStatusPublish，其类型为PublishSubject，它在网络状态发生变化时通知订阅者。这里须要注册一个广播，在收到广播以后，咱们经过mNetStatusPublish通知订阅者，代码以下：

private void registerBroadcast() {
        mReceiver = new BroadcastReceiver() {

            @Override
            public void onReceive(Context context, Intent intent) {
                if (mNetStatusPublish != null) {
                    mNetStatusPublish.onNext(isNetworkConnected());
                }
            }

        };
        IntentFilter filter = new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION);
        registerReceiver(mReceiver, filter);
    }
复制代码

在收到网络状态变化的消息以后：

private Observable<Long> getNetStatusPublish() {
        return mNetStatusPublish.filter(new Predicate<Boolean>() {

            @Override
            public boolean test(Boolean aBoolean) throws Exception {
                return aBoolean && getCacheCity() > 0;
            }

        }).map(new Function<Boolean, Long>() {

            @Override
            public Long apply(Boolean aBoolean) throws Exception {
                return getCacheCity();
            }
            
        }).subscribeOn(Schedulers.io());
    }
复制代码

这里咱们作了两步处理：

• 使用filter对消息进行过滤，只有在 联网状况而且以前已经定位到了城市 以后才通知订阅者，filter的原理图以下所示，该操做符用于过滤掉一些不须要的数据：

  

• 使用map，读取当前缓存的城市名，返回给订阅者，map的原理图以下所示，该操做符能够用于执行变换操做。

  

2.3 网络请求模块

在2.1和2.2中，咱们分别用getCityPublish()和getNetStatusPublish()来获取被订阅者，它们分别对应于定位模块和网络状态模块发生变化时所发送的城市数据，下面来看咱们经过城市数据获取城市天气信息的代码：

private void startUpdateWeather() {
        Observable.merge(getCityPublish(), getNetStatusPublish()).flatMap(new Function<Long, ObservableSource<WeatherEntity>>() {

            @Override
            public ObservableSource<WeatherEntity> apply(Long aLong) throws Exception {
                Log.d(TAG, "尝试请求天气信息=" + aLong);
                return getWeather(aLong).subscribeOn(Schedulers.io());
            }

        }).retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {

            @Override
            public ObservableSource<?> apply(Observable<Throwable> throwableObservable) throws Exception {
                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {

                    @Override
                    public ObservableSource<?> apply(Throwable throwable) throws Exception {
                        Log.d(TAG, "请求天气信息过程当中发生错误，进行重订阅");
                        return Observable.just(0);
                    }

                });
            }

        }).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<WeatherEntity>() {

            @Override
            public void onSubscribe(Disposable disposable) {
                mCompositeDisposable.add(disposable);
            }

            @Override
            public void onNext(WeatherEntity weatherEntity) {
                WeatherEntity.WeatherInfo info = weatherEntity.getWeatherinfo();
                if (info != null) {
                    Log.d(TAG, "尝试请求天气信息成功");
                    StringBuilder builder = new StringBuilder();
                    builder.append("城市名：").append(info.getCity()).append("\n").append("温度：").append(info.getTemp()).append("\n").append("风向：").append(info.getWD()).append("\n").append("风速：").append(info.getWS()).append("\n");
                    mTvNetworkResult.setText(builder.toString());
                }
            }

            @Override
            public void onError(Throwable throwable) {
                Log.d(TAG, "尝试请求天气信息失败");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "尝试请求天气信息结束");
            }
        });
    }

    private Observable<WeatherEntity> getWeather(long cityId) {
        WeatherApi api = new Retrofit.Builder()
                .baseUrl("http://www.weather.com.cn/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
                .build().create(WeatherApi.class);
        return api.getWeather(cityId);
    }
复制代码

这里咱们作了如下几个操做：

• 使用merge合并两个数据源，咱们经过getWeather(long cityId)来获取城市信息，这里面用到了 RxJava2 实战知识梳理(4) - 结合 Retrofit 请求新闻资讯 的知识，只不过这里的接口是使用的天气信息网的数据，merge的原理在 RxJava2 实战知识梳理(8) - 使用 publish + merge 优化先加载缓存，再读取网络数据的请求过程 也已经作了介绍。

• 使用retryWhen进行重订阅，由于在获取到城市，以后转换成城市天气信息的时候有可能发生错误，若是发生了错误，那么整个调用链就结束了，须要从新订阅。这里的重订阅使用的retryWhen操做符，关于重订阅更详细的解释能够看前面的这篇文章 RxJava2 实战知识梳理(6) - 基于错误类型的重试请求，下面是其中的部分说明：

  

• 使用observeOn切换到主线程进行界面的更新，原理如： RxJava2 实战知识梳理(1) - 后台执行耗时操做，实时通知 UI 更新

2.4 示例演示

本章的示例代码在 RxSample 的第十一章中，咱们演示两种状况：

• 正常联网状况，定位回调的间隔为1s：

  

  控制台输出以下，能够看到只有当先后两次定位信息不一样时才会发起网络请求天气信息：

  

• 在非联网的时候进入，并只进行一次定位，而后在切换到有网的状态。
  此时控制台的输出以下，能够看到在网络重连以后，咱们使用缓存的城市自动从新发起了请求：

  

2.6 操做符

在这个示例中，咱们用到了如下几种操做符，若是有不明白的地方，你们能够去对应的连接中查看更详细的解释：

• PublishSubject
• DistinctUntilChanged
• Filter
• Merge
• Map
• SubscribeOn
• ObserveOn

---

更多文章，欢迎访问个人 Android 知识梳理系列：

• Android 知识梳理目录：www.jianshu.com/p/fd82d1899…
• 我的主页：lizejun.cn
• 我的知识总结目录：lizejun.cn/categories/