Android RxJava使用教程_android rajava

Rxjava现在真的很火，看了GitHub上一些项目的源码，发现好多都用了这个框架，那么这个框架有什么用呢，实话说，我现在对这个框架的原理还只是一知半解，只能把我自己最近看的一些资料根据自己的心得总结一下，然后把RxJava的用法用一个简易的Demo进行说明。

RxJava是一个异步处理的库。如果我这么说你还是不懂的话，那么你可以联系我们平时使用的AsyncTask和Handler，它们是Android提供的原生工具，专门进行异步消息处理的。

举个栗子，我们经常需要从网上下载数据然后显示出来（例如显示在TextView上），通常是新开线程进行网络请求，然后在UI线程更新UI，那么使用Handler的话，我们通常这样做：

        new Thread(new Runnable() {
            @Override
            public void run() {

                //进行网络请求操作
                // doSomething...

                //使用handler通知UI线程进行UI更新
                handler.sendEmptyMessage(1);
            }
        }).start();
1
2
3
4
5
6
7
8
9
10
11

如果使用AsyncTask，一般在onPreExecute()方法显示进度条，在doInBackground()进行耗时操作，也就是发起网络请求，然后在onProgressUpdate()中更新下载进度，当数据下载完毕后，调用onPostExecute()方法进行一些收尾工作，例如关闭进度条。AsyncTask本质上也是新开线程进行网络请求，然后在UI线程进行UI更新的，只不过它作了进一步的封装而已。

因此，异步操作是相对于同步操作而言的，同步操作就类似方法调用，方法a调用方法b，必须等方法b执行完毕并返回后才能进行自己的后续操作；而异步操作类似UI线程启动新线程执行耗时操作，新线程执行期间并不会阻塞UI线程，UI线程还是可以响应用户的点击操作，新线程执行完毕后传递一个消息给UI线程即可。

RxJava的工作原理也像AsyncTask，Handler一样，只不过它的逻辑更简单，代码量更少，阅读性更强，在单元测试、代码调试方面具有更明显的优势。

懂了什么叫异步处理之后，我们再来简单看看RxJava的原理，它是基于观察者模式，所谓观察者模式，就是存在一个被观察者，然后其他所有观察者的行为都基于被观察者的行为，通常是当被观察者进行某些操作后，一一通知观察者，我这么说可能很难懂，下面是一些简单的代码：

   public void onClick(View view){
        Obserable obserable=new Obserable(new Subscriber1(),new Subscriber2());
        obserable.inform();
    }


    //被观察者
    class Obserable{

        Subscriber1 subscriber1;
        Subscriber2 subscriber2;

        public Obserable(Subscriber1 subscriber1,Subscriber2 subscriber2){
            this.subscriber1=subscriber1;
            this.subscriber2=subscriber2;
        }

        public void inform(){
            //执行某些操作...

            //通知观察者
            subscriber1.doSomething();
            subscriber2.doSomething();

        }

    }

    //观察者
    class Subscriber1{
        void doSomething(){
            //执行某些操作
        }
    }

    class Subscriber2{
        void doSomething(){
            //执行某些操作
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

这段代码应该很好懂了，就是被观察者observable通过构造函数依赖注入两个被观察者subscriber1和subscriber2，subscriber1和subscriber2并不需要使用线程阻塞的方式等待observable的状态发生改变，而是由observable在自己状态发生改变的时候主动通知两个观察者，通知的方式是调用两个观察者的doSomething()方法。如果还是不懂的建议去找找资料看看观察者模式。

RxJava就是基于观察者模式的，它提供一个Observable类，即被观察者，和一个Subscriber类，即观察者。当观察者订阅被观察者时，被观察者就立即执行特定操作然后通知观察者。

定义一个Subscriber对象需要重写三个回调方法，onNext()、onCompleted()、onError()。
onNext():由被观察者Observable调用，表示一次通知，即被观察者Observable的状态发生改变了，观察者Subscriber需要进行的操作。被观察者可以调用多次onNext()，表示传递多个通知。

onCompleted():由被观察者Observable调用，当被观察者不再调用onNext()，即不再有消息通知时调用该方法。

onError():由被观察者Observable调用，当调用过程中出现异常时，会自动触发这个方法的调用，之后的消息传递会被终止。

如下时一个简单的Subscriber对象，其中的泛型参数指定为String，表示onNext()接收的参数类型。

        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onNext(String s) {
                Toast.makeText(MainActivity.this, s, Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onCompleted() {
                Toast.makeText(MainActivity.this, "Completed!", Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onError(Throwable e) {
                Toast.makeText(MainActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
            }
        };
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

定义一个Observable对象需要重写call()方法，该方法接收一个Subscriber参数，该subcriber表示订阅observable的观察者，我们可以在call方法中通过调用subscriber.onNext()来达到通知观察者的目的。注意：定义了Observable对象并不会调用call()，只有当sucriber订阅了observable后call方法才会被调用。代码如下：

        Observable<String> observable=Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                //执行某些操作...

                //执行完毕后通知被观察者subscriber
                subscriber.onNext("123");
                subscriber.onCompleted();
            }
        });
1
2
3
4
5
6
7
8
9
10

最后，把观察者和被观察者关联起来，即观察者订阅被观察者，当observable调用subscriber()时，会马上调用自己的call()方法。

        observable.subscribe(subscriber);
1

下面我以加载一张图片显示在ImageView作为例子，讲讲RxJava的简单应用。

首先在app-build.gradle中添加rxandroid和rxjava依赖，由于后面会用到Gson解析Json，因此也添加上吧：

dependencies {
    compile fileTree(include: ['*.jar'], dir: 'libs')
    androidTestCompile('com.android.support.test.espresso:espresso-core:2.2.2', {
        exclude group: 'com.android.support', module: 'support-annotations'
    })
    compile 'com.android.support:appcompat-v7:25.3.1'
    testCompile 'junit:junit:4.12'

    //添加rxadroid依赖
    compile 'io.reactivex:rxandroid:1.2.1'
    //添加rxjava依赖
    compile 'io.reactivex:rxjava:1.3.0'
    //添加Gson依赖
    compile 'com.google.code.gson:gson:2.8.1'
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

由于后面有个例子是需要用到网络权限，以防万一，还是现在Manifest.xml中添加上吧：

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

（1）版本1
先上代码：

    @Override
    public void onClick(View view) {
        switch (view.getId()){
            case R.id.btn_show_photo:
                showPhoto1(R.mipmap.ic_launcher);
                break;
        }
    }

   /**
     * RxJava最基本使用方式
     * 加载并展示一张图片
     *
     * @param resId
     */
    private void showPhoto1(final int resId) {
        //观察者
        Subscriber<Drawable> subscriber = new Subscriber<Drawable>() {
            //当被观察者发来消息时所执行的操作
            //该函数由被观察者调用
            @Override
            public void onNext(Drawable d) {
                iv.setImageDrawable(d);
            }

            @Override
            public void onCompleted() {
                Toast.makeText(MainActivity.this, "加载完成", Toast.LENGTH_LONG).show();
            }

            @Override
            public void onError(Throwable e) {

            }
        };

        //被观察者
        Observable<Drawable> observable = Observable.create(new Observable.OnSubscribe<Drawable>() {
            /**
             * 当有观察者订阅被观察者时，call函数会马上被调用
             * @param subscriber
             */
            @Override
            public void call(Subscriber<? super Drawable> subscriber) {
                Drawable drawable = ContextCompat.getDrawable(MainActivity.this, resId);
                //通知观察者
                subscriber.onNext(drawable);
                subscriber.onCompleted();
            }
        });

        observable.subscribe(subscriber);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

首先，我们定义一个观察者subscriber，subscriber需要重写三个方法，onNext()、onCompleted()、onError()，其中onNext()接收Drawable对象作为参数，当接收到被观察者Observable的消息（即Observable调用了subscriber的onNext()方法），观察者就更新ImageView。当通知完毕后，弹出一个Toast提示；
然后我们定义了一个观察者Observable对象并重写了其中的call方法，call()方法中根据resId加载图片并生成对应的Drawable对象后通过调用subscriber的onNext(Drawable)通知观察者。通知完毕后调用onCompleted()表示后续不会再有消息通知了。

点击显示图片按钮后，运行结果如下（EditText和最下方的查询按钮先忽略，后面会用到）：

运行结果

（2）版本2
上面我们分别定义了Observable和Subscriber对象，并将两者通过subscribe()关联起来，其实看看代码量不少，阅读性也不是很好，我们可以把上面的代码简化一下，于是就有了以下的版本2的showPhoto2()函数：

    @Override
    public void onClick(View view) {
        switch (view.getId()){
            case R.id.btn_show_photo:
//                showPhoto1(R.mipmap.ic_launcher);
                showPhoto2(R.mipmap.ic_launcher);
                break;
        }
    }
    /**
     * RxJava使用方式2
     * 加载并展示一张图片
     *
     * @param resId
     */
    private void showPhoto2(final int resId) {
        //使用Observable.just()来生成Observable实例    

        Action1<Drawable> onNextAction = new Action1<Drawable>() {
            @Override
            public void call(Drawable drawable) {
                iv.setImageDrawable(drawable);
            }
        };

        Action0 onCompeletedAction = new Action0() {
            @Override
            public void call() {
                Toast.makeText(MainActivity.this, "显示完毕", Toast.LENGTH_SHORT).show();
            }
        };

        Action1<Throwable> onErrorAction = new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                Toast.makeText(MainActivity.this, throwable.toString(), Toast.LENGTH_SHORT).show();
            }
        };

        Observable.just(ContextCompat.getDrawable(this, resId)).subscribe(onNextAction, onErrorAction, onCompeletedAction);

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

上面我们没有显示定义被观察者Observable对象，而是使用Observable.just(T t)方法，该方法返回一个Observable对象，其参数的类型表示与其订阅的观察者需要接受的参数的类型，在这里就是Drawable。
我们同样没显示定义Subscriber对象，而是定义了Action1和Action0接口的三个对象：onNextAction、onCompletedAction和onErrorAction，想必你从名字就可以看出来了，它们分别对应Subscriber的三个方法：onNext()、onCompleted()和onError()。其中Action1的call方法需要参数，因此可以对应onNext()和onError()，注意onError方法的参数类型是Throwable；onCompleted()不需要参数，可以使用Action0定义。虽然写法跟版本1不一样了，但是两种写法的调用过程和原理是没有区别的。

observable.subscribe()方法有三个重载版本，要注意参数顺序不能颠倒：

// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);
1
2
3
4
5
6

（3）版本3
记得我们一开始说过RxJava是一个异步处理的库，但是我很遗憾的告诉你，我们上两个版本的写法完全不满足异步处理的性质：observable加载图片和iv.setImageDrawable()的操作实际都是在UI线程执行的，也就是说Observable的call方法和Subscriber的onNext()、onCompleted()和onError()方法都是在UI线程执行的。而我们平时一般的做法是把耗时操作放在新线程，UI更新放在UI线程，那么使用RxJava要怎么做呢？方法就是我们版本3的showPhoto3()了！代码如下:

       @Override
    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.btn_show_photo:
//                showPhoto1(R.mipmap.ic_launcher);
//                showPhoto2(R.mipmap.ic_launcher);
                showPhoto3(R.mipmap.ic_launcher);

                break;
        }
    }

    /**
     * 在不同线程执行
     * @param resId
     */
    private void showPhoto3(int resId) {
        Observable.just(ContextCompat.getDrawable(this, resId))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Drawable>() {
                    @Override
                    public void call(Drawable drawable) {
                        iv.setImageDrawable(drawable);
                    }
                });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

我们使用subscribeOn(Schedulers.io())指定被观察者的逻辑都在新线程中执行；使用observeOn(AndroidSchedulers.mainThread())指定观察者的逻辑都在UI线程中执行，这下子就是真正的异步操作啦！

其实到这里你应该是懂得了RxJava最最基本的使用方法了，但是RxJava有一个逆天的功能，那就是RxJava操作符map()的应用啦，这个函数能在消息传递过程中转换对象的格式，我这么说你肯定不懂的了，下面我以手机归属地查询API作为例子来讲解一下。

我们使用一个手机归属地查询API：http://cx.shouji.360.cn/phonearea.php?number=13888888888

使用百度的API调试工具结果如下：

查询结果

可见返回的是Json数据，这里我们使用Gson来解析，不会使用Json的朋友可以看看我这篇文章：
Android 使用Gson解析Json和Json数组 一眼就能快速上手

我创建了一个PhoneResponse类表示返回的Json结果，Json结果中有个”data”的键，其对应的也是个Json，因此还需要创建一个PhoneData类。

PhoneData.java:

public class PhoneData {
    private String province;
    private String city;
    private String sp;

    public void setProvince(String province) {
        this.province = province;
    }

    public String getProvince() {
        return province;
    }

    public void setCity(String city) {
        this.city = city;
    }

    public String getCity() {
        return city;
    }

    public void setSp(String sp) {
        this.sp = sp;
    }

    public String getSp() {
        return sp;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

PhoneResponse.java:

public class PhoneResponse {
    private int code;
    private PhoneData data;

    public void setCode(int code) {
        this.code = code;
    }

    public void setData(PhoneData data) {
        this.data = data;
    }

    public int getCode() {
        return code;
    }

    public PhoneData getData() {
        return data;
    }

    @Override
    public String toString() {
        StringBuilder sb=new StringBuilder();
        sb.append(data.getProvince()).append(" ").append(data.getCity());

        return sb.toString();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

现在先简单说说API查询的过程，这个过程涉及到数据格式的转换，好对应我们等会使用的map操作符。首先是获取EditText中的手机号码（String），然后发起网络请求，结果返回一个Json字符串（String），然后把Json字符串转换为Java对象（PhoneResponse），然后获取PhoneRespon中的PhoneData对象并显示手机归属地到TextView中。

因此，转换流程如下：手机号码（String）->Json字符串（String）->PhoneResponse对象，最后subscriber获取到PhoneResponse对象，从中取出手机归属地信息并显示在TextView中。

手机号码（String）->Json字符串（String）的代码如下：

    public static final String QUERY_URL = "http://cx.shouji.360.cn/phonearea.php?number=";
    /**
     * 使用EditText的手机号进行网络请求
     * @param phoneNum 手机号码
     * @return Json字符串
     *
     */
    private String getJson(String phoneNum) {
        HttpURLConnection connection = null;
        StringBuilder json = new StringBuilder();

        try {
            URL url = new URL(QUERY_URL + phoneNum.trim());
            connection = (HttpURLConnection) url.openConnection();

            BufferedReader reader = new BufferedReader(new InputStreamReader(connection.getInputStream()));
            String line = null;
            while ((line = reader.readLine()) != null) {
                json.append(line);
            }

        } catch (MalformedURLException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }


        return json.toString();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

Json字符串（String）->PhoneResponse对象的代码如下：

    /**
     * 把网络请求返回的Json字符串转换为PhoneResponse对象
     * @param json
     * @return
     */
    private PhoneResponse getPhoneResponse(String json) {
        PhoneResponse phoneResponse = new Gson().fromJson(json, PhoneResponse.class);
        return phoneResponse;
    }
1
2
3
4
5
6
7
8
9

然后转换过程交给RxJava来做，代码如下：

    @Override
    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.btn_show_photo:
//                showPhoto1(R.mipmap.ic_launcher);
//                showPhoto2(R.mipmap.ic_launcher);
                showPhoto3(R.mipmap.ic_launcher);
                break;
            case R.id.btn_query:
                showPhonePlace(et.getText().toString().trim());
                break;
        }
    }

    /**
     * RxJava进行转换：手机号码（String）->Json字符串（String）->PhoneResponse对象
     *
     * 观察者获取到PhoneResponse对象后获取手机归属地信息后显示子TextView中
     * @param phoneNum
     */
    private void showPhonePlace(final String phoneNum) {
        Observable.just(phoneNum)
                //手机号码phoneNum（String）->Json字符串（String）
                .map(new Func1<String, String>() {
                    @Override
                    public String call(String s) {
                        return getJson(s);
                    }
                })
                //Json字符串（String）->PhoneResponse对象
                .map(new Func1<String, PhoneResponse>() {
                    @Override
                    public PhoneResponse call(String json) {
                        return getPhoneResponse(json);
                    }
                })
                //网络请求是耗时操作，放在新线程中处理
                .subscribeOn(Schedulers.io())
                //更新UI必须在UI线程中
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<PhoneResponse>() {
                    @Override
                    public void call(PhoneResponse phoneResponse) {
                        //显示归属地到TextView中
                        tv.setText(phoneResponse.toString());
                    }
                });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

注释已经很清晰了，我说一下map()，map函数接受一个Func1的接口参数，Func1

//手机号码phoneNum（String）->Json字符串（String）
observable.map(new Func1<String, String>() {
                    @Override
                    public String call(String s) {
                        return getJson(s);
                    }
                })
1
2
3
4
5
6
7

像下面的这行代码，就是String->PhoneResponse对象了：

            //Json字符串（String）->PhoneResponse对象
             observable.map(new Func1<String, PhoneResponse>() {
                    @Override
                    public PhoneResponse call(String json) {
                        return getPhoneResponse(json);
                    }
                })
1
2
3
4
5
6
7

运行结果如下：

<script type="math/tex" id="MathJax-Element-3"> </script>
运行结果

map操作符就介绍到了这里了，RxJava还有许多功能强大的操作符，大家可以好好研究研究。Retrofit也是火的不行，所以我就再写一个RxJava+Retrofit2的例子吧。

我以前写过一篇非常简单的Retorfit2教程，还不会Retrofit2的朋友可以先看看：
Android Retrofit2使用教程-小白篇

首先老规矩，添加依赖，还是在app-build.gradle中添加，以下添加的三个依赖的版本必须一致：

    //以下两个是retrofit2所需要的依赖
    compile 'com.squareup.retrofit2:retrofit:2.3.0'
    compile 'com.squareup.retrofit2:converter-gson:2.3.0'

    //rxjava+retrofit2配合使用的依赖
    compile 'com.squareup.retrofit2:adapter-rxjava:2.3.0'
1
2
3
4
5
6

然后创建Http请求的接口，注意函数的返回值是Observable而不是Call<T>
PhoneService.java：

public interface PhoneService {

    //注意返回值是Observable
    @GET("phonearea.php")
    Observable<PhoneResponse> getPhoneResponse(@Query("number") String phoneNum);
}
1
2
3
4
5
6

然后使用Retrofit2发起网络请求：

    @Override
    public void onClick(View view) {
        switch (view.getId()) {
            case R.id.btn_show_photo:
//                showPhoto1(R.mipmap.ic_launcher);
//                showPhoto2(R.mipmap.ic_launcher);
                showPhoto3(R.mipmap.ic_launcher);
                break;
            case R.id.btn_query:
//                showPhonePlace(et.getText().toString().trim());
                showPhonePlaceWithRetrofit(et.getText().toString().trim());
                break;
        }
    }

   /**
     * rxjava+retrofit实例
     * @param phoneNum
     */
    private void showPhonePlaceWithRetrofit(final String phoneNum){
        Retrofit retrofit=new Retrofit.Builder()
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) //这一行是要新添加上的
                .baseUrl("http://cx.shouji.360.cn")
                .build();

        PhoneService service=retrofit.create(PhoneService.class);

        //后面的就基本是一样的了
        service.getPhoneResponse(phoneNum)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<PhoneResponse>() {
                    @Override
                    public void onCompleted() {
                        Toast.makeText(MainActivity.this, "查询成功", Toast.LENGTH_SHORT).show();
                    }

                    @Override
                   public void onError(Throwable e) {
                        Toast.makeText(MainActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
                    }

                    @Override
                    public void onNext(PhoneResponse phoneResponse) {
                        tv.setText(phoneResponse.toString());
                    }
                });
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

最后运行结果是一样的。

对RxJava的一个简单讲解就到这里啦，最后奉上源码，希望对你能有所帮助！
源码下载