RxJava入门详解

RxJava火了一年多了，借着上次在组内分享的机会，好好学习研究了一把。在此把学习成果记录分享一下，分为入门和深入两部分。

一、RxJava是什么？有什么用？

a library for composing asynchronous and event-based programs using observable sequences for the Java VM。（一个在Java VM上使用可观测的序列来组成异步的、基于事件的程序的库）

（这句话看了跟没看似的，只get到异步两个字）看不懂没关系，继续~~~

那RxJava有什么用呢？

简洁：它能使代码在逻辑变得越来越复杂的情况下依然保持简洁。（大家都这么说~~）

功能强大：它提供了丰富的操作符，方便做各种数据处理操作。（熟练的掌握操作符，才能发挥RxJava的威力呀！！）

二、基本原理

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

观察者模式

观察者模式是一种经常使用的设计模式，它包含两个重要角色：观察者(Observer)和被观察者(Observable)。一个Observer对象监视着一个Observable对象的变化，当Observable对象发生变化时，Observer得到通知，就可以进行相应的工作。

RxJava中的观察者模式

RxJava的观察者模式中有四个基本概念：
Observable (被观察者)、Observer (观察者)、subscribe (订阅)、事件。Observable和Observer通过subscribe()方法实现订阅关系，从而Observable可以在需要的时候发出事件来通知Observer。

RxJava中的观察者模式与传统的观察者模式略有不同，它有三个事件回调方法，分别是：
•onNext()：相当于传统观察者模式的onEvent()。
•onCompleted()：事件完结队列。RxJava规定，当不会再有新的onNext()发出时，需要触发onCompleted()方法作为标志。
•onError()：事件异常队列。在事件处理过程中出异常时，onError()会被触发，同时队列自动终止，不允许再有事件发出。

说了那么多，先看一段代码感受一下吧~~~

//第一步：
// 创建被观察者
Observable observable= Observable.from(newString[]{"Hello","world!"});

//第二步：
// 使用Subscriber做为观察者,Subscriber对Observer接口进行了一些扩展
Subscriber subscriber=new Subscriber() {

@Override
public void onCompleted() {
     Log.i(TAG,"onCompleted ! ");
}

@Override
public void onError(Throwable e) {
    Log.i(TAG,"onError ! ");
}

@Override
public void onNext(String s) {
    Log.i(TAG,"onNext, s = "+ s);
}
};

// 使用Observer做为观察者
Observer observer=new Observer() {

@Override
public void onCompleted() {
    Log.i(TAG,"onCompleted ! ");
}

@Override
public void onError(Throwable e) {
    Log.i(TAG,"onError ! ");
}

@Override
public void onNext(String s) {
    Log.i(TAG,"onNext, s = "+ s);
}
};

//第三步：
//订阅
observable.subscribe(subscriber); //订阅方法一
// or
observable.subscribe(observer); //订阅方法二

上面的代码主要有三步：1）创建创建被观察者；2）创建观察者；3）建立被观察者与观察者之间的订阅关系。
其中，分别演示使用了Observer和Subscriber作为观察者。Observer和Subscriber之间的区别是：Subscriber实现了Observer接口，比Observer多了一个最重要的方法unsubscribe( )，用来取消订阅，当你不再想接收数据了，可以调用unsubscribe( )方法停止接收。Observer在subscribe() 过程中，最终也会被转换成Subscriber对象，一般情况下，建议使用Subscriber作为接收源。

上面两种订阅方法运行的结果是相同的。分别会调用两次onNext()方法，调用一次onComplete()方法。运行如果如下：

02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onNext, s = Hello
02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onNext, s = world!
02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onCompleted ! 
02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onNext, s = Hello
02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onNext, s = world!
02-04 16:14:26.467 2795-2795/com.rx.demo I/MainActivity: onCompleted ! 

三、线程控制

在不指定线程的情况下，RxJava遵循的是线程不变的原则。即：在哪个线程调用subscribe()，就在哪个线程生产事件；在哪个线程生产事件，就在哪个线程消费事件。如果需要切换线程，就需要用到Scheduler(调度器)。

线程控制的两个重要方法：
subscribeOn(scheduler) ：主要改变的是订阅的线程，即call()执行的线程。它作用于该操作符之前的Observable的创建操作符以及doOnSubscribe操作符;
observeOn(scheduler)：主要改变的是发送的线程，即onNext()执行的线程。它作用于该操作符之后的操作符直到出现新的observeOn操作符。

还是先用代码感受一下吧！

final int drawableRes = R.mipmap.image_01;
        final ImageView imageView = (ImageView) findViewById(R.id.iv_image);
        Observable.create(
                new Observable.OnSubscribe<Drawable>() {
                    @Override
                    public void call(Subscriber<? super Drawable> subscriber) {
                        Drawable drawable = getTheme().getDrawable(drawableRes);
                        subscriber.onNext(drawable);
                        subscriber.onCompleted();
                        Log.e(TAG, "call current Thread #" + android.os.Process.myTid());
                    }
                })
                .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
                .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程
                .subscribe(new Observer<Drawable>() {
                    @Override
                    public void onNext(Drawable drawable) {
                        imageView.setImageDrawable(drawable);
                        Log.e(TAG, "onNext current Thread #" + android.os.Process.myTid());
                    }

                    @Override
                    public void onCompleted() {
                    }

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

线程切换日志：

02-04 16:36:32.205 2795-2795/com.rx.demo E/MainActivity: onCreate current Thread #2795
02-04 16:36:32.207 2795-24441/com.rx.demo E/MainActivity: call current Thread #24441
02-04 16:36:32.216 2795-2795/com.rx.demo E/MainActivity: onNext current Thread #2795

上面的代码通过subscribeOn(Schedulers.io())，将加载图片的操作切换到IO线程，然后observeOn(AndroidSchedulers.mainThread())将Subscriber的回调方法切换回主线程，将图片显示在ImageView上，避免了加载图片过程中造成界面卡顿。

RxJava内置的几个Scheduler：
Schedulers.immediate(): 直接在当前线程运行，相当于不指定线程(默认)。
Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。
Schedulers.io(): I/O 操作(读写文件、读写数据库、网络信息交互等)所使用的Scheduler。行为模式和newThread()差不多，区别在于io()的内部实现是是用一个无数量上限的线程池，可以重用空闲的线程，因此多数情况下io()比newThread()更有效率。不要把计算工作放在io()中，可以避免创建不必要的线程。
Schedulers.computation(): 计算所使用的Scheduler。这个计算指的是CPU密集型计算，即不会被I/O等操作限制性能的操作，例如图形的计算。这个Scheduler使用的固定的线程池，大小为CPU核数。不要把I/O操作放在computation()中，否则I/O操作的等待时间会浪费CPU。
AndroidSchedulers.mainThread()：Android专有，它指定的操作将在Android主线程运行。

看完之后，是不是感觉RxJava的线程控制非常简单呀~~~

四、RxJava的变换

所谓变换，就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列。
RxJava提供了对事件序列进行变换的支持，这是它的核心功能之一。这是“RxJava真是太好用了”的最大原因。

RxJava的变换是通过操作符(Operators)来实现的。RxJava内置了大量的操作符用于对数据的变化处理操作，下面简单列举了一些操作，点击更多操作

转换操作：用于对Observable发射的数据进行变换;
过滤操作：用于从Observable发射的数据中进行选择;
组合操作：用于将多个Observable组合成一个单一的Observable;
条件和布尔操作：用于单个或多个数据项，也可用于Observable;
算术和聚合操作：可用于整个数据序列;
连接操作：
…

同样的用代码感受一下操作符的作用吧~~~
假设我们需要查看若干个学生分别选修了什么课程。即有一个Student的数组：

Student[] students = new Student[]{
    new Student("张三", new Course[]{
        new Course("语文"),
        new Course("数学")
    }),
    new Student("李四", new Course[]{
        new Course("物理"),
        new Course("化学")
    }),
    new Student("王五", new Course[]{
        new Course("英语"),
        new Course("计算机")
    }),
};

使用传统方法，需要使用两次for循环来遍历：

for (Student student : students){
   for (Course course : student.courses){
      Log.e(TAG, " course name = " + course.name);
   }
}

使用RxJava的flatMap操作符：

private void doFlatMap() {
        Subscriber<Course> subscriber = new Subscriber<Course>() {
            @Override
            public void onNext(Course course) {
                Log.e(TAG, " course name : " + course.name);
            }

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
            }
        };

        Observable.from(students) // students是一个Student[]对象
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        //使用传入的事件对象创建一个Observable对象；
                        return Observable.from(student.courses);
                    }
                })
                .subscribe(subscriber);
}

//输出Log
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 语文
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 数学
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 物理
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 化学
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 英语
02-04 17:16:14.815 29307-29307/com.rx.demo E/MainActivity:  course name : 计算机

对比两种方法，使用RxJava似乎不占优势，但是它让代码保持简洁。

当然了，RxJava还有更多功能强大的操作符，大家可以慢慢挖掘。

本篇博客主要是介绍RxJava的入门使用。
下一篇博客介绍RxJava是如何实现异步以及如何实现数据变换？。

欢迎大家评论交流~~~