正文

Android RxJava/RxAndroid结合Retrofit使用

IP云
IP云V管理员/0 阅读
11/28
小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

概述

RxJava是一个在 Java VM 上使用可观測的序列来组成异步的、基于事件的程序的库。更重要的是：使用RxJava在代码逻辑上会非常简洁明了，尤其是在复杂的逻辑上。告别迷之缩进。

RxAndroid是RxJava针对Android平台的拓展。

Retrofit是一个封装了okHttp的工具库。在上篇博文 Android 初探Retrofit2.0.1（最新版） 有过介绍，对Retrofit不太了解的读者，建议先行阅读。

RxJava GitHub地址：https://github.com/ReactiveX/RxJava

RxAndroid GitHub地址：https://github.com/ReactiveX/RxAndroid

Retrofit GitHub地址：https://github.com/square/retrofit

热身运动 - 观察者模式

RxJava 的异步实现，是通过一种扩展的观察者模式来实现的。

以下简介下观察者模式。熟练掌握观察者模式可跳过这一小节。

观察者模式

如果如今有两个对象A和B。在A发生某种变化时要主动通知B。这就是观察者模式。

Android里 View.setOnClickListener(new OnClickListener) 就是运用观察者模式的典型样例。在这个样例中View充当对象A的角色，OnClickListener充当B。View通过setOnClickListener将自己和OnClickListener联系（订阅）起来。当View捕获到点击事件之后，立刻调用 OnClickListener#onClick() 方法。还有通常我们自定义的接口回调都是观察者模式的运用。

RxJava的观察者模式

RxJava基本概念：Observable (被观察者，相当于View)、 Observer (观察者，相当于OnClickListener)、 subscribe ()(订阅。相当于setOnClickListener()方法)事件。Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 能够在须要的时候发出事件来通知 Observer。

RxJava除了普通的回调方法onNext()还有onCompleted() 和 onError()。

  1. onCompleted()：事件队列完结。

    RxJava 不仅把每一个事件单独处理。还会把它们看做一个队列。没有新的onNext()之后，调用此方法。

  2. onError()：事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同一时候队列自己主动终止，不同意再有事件发出。
  3. onCompleted() 和 onError()在一个队列中仅仅能调用一个，而且是最后一个。onCompleted() 和 onError()还是相互排斥的，仅仅能调用当中一个

回想Retrofit

上篇博文我们使用Retrofit实现对网络的訪问以及返回数据的解析，详情请见Android 初探Retrofit2.0.1（最新版），在这里我们再简单温习下

1. 创建WeatherInfoService，并制定请求数据的方式以及须要的查询參数

2. 创建对应的WeatherInfoBean

3. 创建Retrofit对象并使用GSON解析数据

4. 调用 weatherInfoService#getWeatherInfo(FORMAT, CITYNAME, KEY)，获取call

5. 插入队列，并展示数据

RxJava/RxAndroid结合Retrofit

加入依赖 

compile 'com.squareup.retrofit2:retrofit:2.0.1'

compile 'com.squareup.retrofit2:converter-gson:2.0.0-beta4'

compile 'com.squareup.retrofit2:adapter-rxjava:2.0.1'

compile 'io.reactivex:rxandroid:1.1.0'

compile 'io.reactivex:rxjava:1.1.0'

加入限权

<uses-permission android:name="android.permission.INTERNET" />

再次见到熟悉的WeatherInfoService 

public interface WeatherInfoService {

    @GET("http://v.juhe.cn/weather/index?format=2&cityname=北京&key=b952ad7acbc7415f3f3c9bf274e39c45")

    Observable<WeatherInfo> getWeatherInfoByRxJava();

}

注意这里 getWeatherInfoByRxJava() 的返回类型为 Observable<WeatherInfo>， 这也就意味着我们在这里直接得到一个被观察者Observable！

还是那个MainActivity 

        findViewById(R.id.btn2).setOnClickListener(new View.OnClickListener() {

            @Override

            public void onClick(View v) {

                Gson gson = new GsonBuilder().create();

                Retrofit retrofit = new Retrofit.Builder()

                        .baseUrl(BASE_URL)

                        //配置转化库，默认是Gson

                        .addConverterFactory(GsonConverterFactory.create(gson))

                        //配置回调库，採用RxJava

                        .addCallAdapterFactory(RxJavaCallAdapterFactory.create())

                        .build();

                WeatherInfoService weatherInfoService = retrofit.create(WeatherInfoService.class);

                getWeatherInfo(weatherInfoService,gson);

                getWeatherInfoByMap(weatherInfoService,gson);

                getWeatherInfoByFlatMap(weatherInfoService,gson);

             }

        });

注意 .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) ，这行代码的作用是配置Retrofit回调库，採用RxJava。

这里我们写了 getWeatherInfo(weatherInfoService,gson); 、 getWeatherInfoByMap(weatherInfoService,gson); 和 getWeatherInfoByFlatMap(weatherInfoService,gson); 三个方法。接下来会一一解说

MainActivity#getWeatherInfo()

在这种方法里实现了最主要的RxJava/RxAndroid和Retrofit的结合。代码例如以下 

        weatherInfoService.getWeatherInfoByRxJava()// 返回类型为Observable<WeatherInfo>

                .subscribeOn(Schedulers.io())// 指定订阅者在io线程（第一次指定订阅者线程有效）

                .doOnSubscribe(new Action0() { //doOnSubscribe线程为近期的subscribeOn指定线程

                    @Override

                    public void call() {

                        Toast.makeText(MainActivity.this, "我在网络请求前运行", Toast.LENGTH_SHORT).show();

                    }

                })

                .subscribeOn(AndroidSchedulers.mainThread())// 指定在主线程

                .observeOn(AndroidSchedulers.mainThread())// 指定观察者在主线程

                .subscribe(new Subscriber<WeatherInfo>() {

                    @Override

                    public void onCompleted() {

                        Log.i(TAG, "onCompleted");

                    }

                    @Override

                    public void onError(Throwable e) {

                        tv.setText("error:" + e.getMessage());

                    }

                    @Override

                    public void onNext(WeatherInfo weatherInfo) {

                        Log.i(TAG, gson.toJson(weatherInfo));

                        tv.setText(gson.toJson(weatherInfo));

                    }

                });

代码里凝视非常多。补充几个须要注意的地方。

1. subscribeOn()仅仅有第一次调用的时候就指定被观察者Observable所在线程。以后能够多次调用。但被观察者Observable所在线程已经指定

2. doOnSubscribe()在发送事件前运行，能够指定运行线程。

一般在里面展示loading

3. observeOn()可多次调用而且每次都会改变观察者Observer/Subscriber所在线程。

MainActivity#getWeatherInfoByMap()

Observable的map()是个奇妙的方法。它能够对被观察者Observable的泛型进行操作，而且返回还有一个Observable传递给观察者Observer/Subscriber 

    private void getWeatherInfoByMap(WeatherInfoService weatherInfoService, final Gson gson) {

        weatherInfoService.getWeatherInfoByRxJava()// 返回类型为Observable<WeatherInfo>

                .subscribeOn(Schedulers.io())// 指定订阅者在io线程（第一次指定订阅者线程有效）

                .doOnSubscribe(new Action0() { //doOnSubscribe线程为近期的subscribeOn指定线程

                    @Override

                    public void call() {

                        Toast.makeText(MainActivity.this, "我在网络请求前运行", Toast.LENGTH_SHORT).show();

                    }

                })

                .subscribeOn(AndroidSchedulers.mainThread())// 指定在主线程

                .map(new Func1<WeatherInfo, Today>() {

                    @Override

                    public Today call(WeatherInfo weatherInfo) {

                        return weatherInfo.getResult().getToday();

                    }

                })

                .observeOn(AndroidSchedulers.mainThread())// 指定观察者在主线程

                .subscribe(new Subscriber<Today>() {

                    ...

                    @Override

                    public void onNext(Today today) {

                        tv.setText(gson.toJson(today));

                    }

                });

    }

在.map()方法中我们获取WeatherInfo中Today属性，而且返回Today。然后再观察者Subscriber我们就能够直接对Today进行操作。是不是非常方便？还有更方便的！

MainActivity#getWeatherInfoByFlatMap()

使用.map方法仅仅能返回一个值，属于一对一类型。

RxJava给我们提供一个更奇妙的方法.flatMap()。 

    private void getWeatherInfoByFlatMap(WeatherInfoService weatherInfoService, final Gson gson) {

        weatherInfoService.getWeatherInfoByRxJava()

                .subscribeOn(Schedulers.io())

                .flatMap(new Func1<WeatherInfo, Observable<Future>>(){

                    @Override

                    public Observable<Future> call(WeatherInfo weatherInfo) {

                        return Observable.from(weatherInfo.getResult().getFuture());

                    }

                })

                .observeOn(AndroidSchedulers.mainThread())

                .subscribe(new Subscriber<Future>() {

                    @Override

                    public void onCompleted() {

                        tv.setText(sb);

                    }

                    @Override

                    public void onError(Throwable e) {

                    }

                    @Override

                    public void onNext(Future future) {

                        Log.i(TAG,gson.toJson(future));

                       sb.append(gson.toJson(future)+"\n");

                    }

                });

    }

这种方法有些难以理解，以下具体介绍关键部分代码。首先在 .flatMap()中 第一个參数为被观察者Observable的泛型WeatherInfo。第二个參数定义为还有一个被观察者，为了叙述方便，下文称第一个被观察者A。第二个參数即还有一个被观察者称为B。A的泛型为WeatherInfo。B的泛型为Future。getFuture是这么被定义的 

public class WeatherInfo {

    private String resultcode;

    private String reason;

    private String error_code;

    private Result result;

    ...

}

public class Result {

    private SK sk;

    private Today today;

    private List<Future> future;

    public List<Future> getFuture() {

        return future;

    }

    ...

}

原来 getFuture() 返回的是个 List<Future> ,但是在Func1的 call() 返回值怎么怎么是 Observable<Future> ？这是由于 Observable.from() 会将 List<Future> 拆分成一个个的Future返回，也就是说订阅者的 onNext 方法将会被运行 List<Future>.seze() 次。所以这里我们定义了一个sb（StringBuilder）。用于将每次返回的Future拼接起来，最后在 onCompleted() 中调用 tv.setText(sb);

结束语

至此，RxJava/RxAndroid结合Retrofit解说完成，希望能对读者有所帮助。 感谢耐心读到最后！

源代码下载地址：0分仅仅为共享

參考：

http://gank.io/post/560e15be2dca930e00da1083

http://blog.csdn.net/liuhongwei123888/article/details/50375897

相关阅读

目录[+]