RxJava2 实战系列文章

RxJava2 实战知识梳理(1) – 后台执行耗时操作,实时通报 UI
更新
RxJava2 实战知识梳理(2) –
计算一段时间内数据的平均值
RxJava2 实战知识梳理(3) –
优化搜索联想功能
RxJava2 实战知识梳理(4) – 结合 Retrofit
请求新闻消息
RxJava2 实战知识梳理(5) –
简单和进阶的轮询操作
RxJava2 实战知识梳理(6) –
基于左类型的重试请求
RxJava2 实战知识梳理(7) – 基于 combineLatest
实现之输入表单验证
RxJava2 实战知识梳理(8) – 使用 publish + merge
优化先加载缓存,再念博网络数据的乞求过程
RxJava2 实战知识梳理(9) – 使用 timer/interval/delay
实现任务调度
RxJava2 实战知识梳理(10) – 屏幕旋转导致 Activity
重建时回升任务
RxJava2 实战知识梳理(11) –
检测网络状态并机关重试请求
RxJava2 实战知识梳理(12) – 实战讲解 publish & replay & share & refCount
& autoConnect
RxJava2 实战知识梳理(13) –
如何让错误产生时未自动停止订阅关系
RxJava2 实战知识梳理(14) – 在 token 过期时,刷新过期 token
并再度发起呼吁
RxJava2 实战知识梳理(15) – 实现一个简约的 MVP + RxJava + Retrofit
应用


同一、应用背景

今日,我们坐一个请求天气数据的例子,来演示如何用RxJava兑现网络又连时的机关请求,首先,我们本着是要求进行一个简的叙说,整个项目的框架如下所示:

新濠娱乐 1

总体框架图

  • 每当采用启动时,我们会启动定位模块,该定位模块于后台每隔一段时间发起一坏定位要,拿到定点的结果后,我们透过该城于服务器发起呼吁,以得对诺都之天气信息进行亮。
  • 然而在用到都后向服务器请求天气的长河中来或是处于无网络的状态,导致无法取得城市之气象信息并刷新界面,因此,我们用检测网络的状态,在网络重连的时段,读取上一致次缓存的都,向服务器发起呼吁以获得城市相应天气信息。

正文的演示代码在
RxSample
的第十一章节中。

二、示例

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的原理图如下所示:

    新濠娱乐 2

  • 使用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的原理图如下所示,该操作符用于过滤掉一部分未需要的数额:

    新濠娱乐 3

  • 使用map,读取当前缓存的都名,返回给订阅者,map的法则图如下所示,该操作符可以用于实践变换操作。

    新濠娱乐 4

2.3 网络要模块

2.12.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) –
    基于左类型的重试请求,下面是中间的一些说明:

    新濠娱乐 5

  • 使用observeOn切换至主线程进行界面的创新,原理如:
    RxJava2 实战知识梳理(1) – 后台执行耗时操作,实时通报 UI
    更新

2.4 示例演示

本章的演示代码在
RxSample
的第十一章节中,我们演示两种情况:

  • 正规联网状态,定位回调的区间也1s

    新濠娱乐 6

控制台输出如下,可以看到只有当前后两次定位信息不同时才会发起网络请求天气信息:  

![](https://upload-images.jianshu.io/upload_images/1949836-4d44a2d024be5556.png)
  • 以非联网的下进,并只进行同样次等定位,然后在切换到有网的状态。
![](https://upload-images.jianshu.io/upload_images/1949836-83e7eb4c9f6252bd.gif)



此时控制台的输出如下,可以看到在网络重连之后,我们使用缓存的城市自动重新发起了请求:



![](https://upload-images.jianshu.io/upload_images/1949836-dbfacb32f0857ed4.png)

2.6 操作符

当此示例中,我们之所以到了以下几种操作符,如果出免明白的地方,大家可错过相应的链接中查阅更详细的解说:

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

又多篇,欢迎访问我的 Android 知识梳理系列:

  • Android
    知识梳理目录:http://www.jianshu.com/p/fd82d18994ce
  • 个人主页:http://lizejun.cn
  • 私家文化总结目录:http://lizejun.cn/categories/
网站地图xml地图