RxJava是ReactiveX中使用Java语言实现的版本，而ReactiveX是一种基于异步数据流概念的编程模式，响应式编程。

RxJava 有四个基本概念：

Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。
Observable 和 Observer 通过 subscribe() 方法实现订阅关系，从而 Observable 可以在需要的时候发出事件来通知 Observer。

• onCompleted(): 事件队列完结。RxJava 不仅把每个事件单独处理，还会把它们看做一个队列。RxJava 规定，当不会再有新的
• onNext() 发出时，需要触发 onCompleted() 方法作为标志。
• onError(): 事件队列异常。在事件处理过程中出异常时，onError() 会被触发，同时队列自动终止，不允许再有事件发出。
• 在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，并且是事件序列中的最后一个。需要注意的是，onCompleted() 和 onError() 二者也是互斥的，即在队列中调用了其中一个，就不应该再调用另一个。

Observer 观察者

它决定事件触发的时候将有怎样的行为。在Observer中有三个方法，onNext()、onCompleted()、onError();

//创建Observer 
Observer<String> observer = new Observer<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }
    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

除了Observer接口之外，RxJava还内置了一个实现了 Observer 的抽象类：Subscriber。Subscriber对Observer接口进行了一些扩展，但他们的基本使用方式是完全一样。

Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }
    
    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};

• onStart(): 这是 Subscriber 增加的方法。它会在 subscribe 刚开始，而事件还未发送之前被调用，可以用于做一些准备工作，例如数据的清零或重置。

  这是一个可选方法，默认情况下它的实现为空。需要注意的是，如果对准备工作的线程有要求（例如弹出一个显示进度的对话框，这必须在主线程执行）， onStart() 就不适用了，因为它总是在 subscribe 所发生的线程被调用，而不能指定线程。要在指定的线程来做准备工作，可以使用 doOnSubscribe() 方法，具体可以在后面的文中看到。

• unsubscribe(): 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。一般在这个方法调用前，可以使用 isUnsubscribed() 先判断一下状态。

  unsubscribe() 这个方法很重要，因为在 subscribe() 之后， Observable 会持有 Subscriber 的引用，这个引用如果不能及时被释放，将有内存泄露的风险。所以最好保持一个原则：要在不再使用的时候尽快在合适的地方（例如 onPause() onStop() 等方法中）调用 unsubscribe() 来解除引用关系，以避免内存泄露的发生。

Observable 被观察者

它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable。

在RxJava中有几种不同的Observables:

• Observable<T>：能够发射0或者n个数据，并以成功或错误事件终止。
• Flowable<T>：能够发射0或者n个数据，并以成功或错误事件终止。支持Backpressure，可以控制数据流发射速度。
• Single<T>：只发射单个数据或错误事件。
• Completable：他从来不发射数据，只处理onComplete和onError事件。可以看成是Rx的Runable。
• Mapbe<T>：能够发射0或者1个数据，要么成功，要么失败，有点类似于Optional

//创建Observable
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onNext("RxJava");
        subscriber.onCompleted();
    }
});

create() 方法是 RxJava 最基本的创造事件序列的方法。基于这个方法， RxJava 还提供了一些方法用来快捷创建事件队列:

• just(T...): 将传入的参数依次发送出来。

Observable observable = Observable.just("Hello", "RxJava");
// 将会依次调用：
// onNext("Hello");
// onNext("RxJava");
// onCompleted();

• from(T[]) / from(Iterable<? extends T>) : 将传入的数组或 Iterable 拆分成具体对象后，依次发送出来。

String[] words = {"Hello", "RxJava"};
Observable observable = Observable.from(words);
// 将会依次调用：
// onNext("Hello");
// onNext("RxJava");
// onCompleted();

Subscribe 订阅

创建了 Observable 和 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。

observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);

Action

Action是RxJava 的一个接口，常用的有Action0和Action1。

• Action0： 它只有一个方法 call()，这个方法是无参无返回值的；

  由于 onCompleted() 方法也是无参无返回值的，因此 Action0 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。

• Action1：它同样只有一个方法 call(T param)，这个方法也无返回值，但有一个参数；

  与 Action0 同理，由于 onNext(T obj) 和 onError(Throwable error) 也是单参数无返回值的，因此 Action1 可以将 onNext(obj)和 onError(error) 打包起来传入 subscribe() 以实现不完整定义的回调

创建被观察者和观察者并订阅，Override时只用了onNext(obj)，但是还有onError(error)和onCompleted()必须被重写，但是这两个方法并没有被用到导致产生无用代码。

Observable.just("Hello", "RxJava")
                .subscribe(new Subscriber<String>() {
                    @Override
                    public void onCompleted() {
                        
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onNext(String s) {
                        Log.i(TAG, s);
                    }
                });
//Action来代替Subscriber
Observable.just("Hello", "RxJava")
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, s);
                    }
                });

定义三个对象，分别打包onNext(obj)、onError(error) 、onCompleted()方法。

Observable observable = Observable.just("Hello", "RxJava");
        //处理onNext()中的内容
        Action1<String> onNextAction = new Action1<String>() {
            @Override
            public void call(String s) {
                Log.i(TAG, s);
            }
        };
        //处理onError()中的内容
        Action1<Throwable> onErrorAction = new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {

            }
        };
        //处理onCompleted()中的内容
        Action0 onCompletedAction = new Action0() {
            @Override
            public void call() {
                Log.i(TAG, "Completed");

            }
        };

使用subscribe重载的方法

//使用 onNextAction 来定义 onNext()
Observable.just("Hello", "RxJava").subscribe(onNextAction);
//使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
Observable.just("Hello", "RxJava").subscribe(onNextAction, onErrorAction);
//使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
Observable.just("Hello", "RxJava").subscribe(onNextAction, onErrorAction, onCompletedAction);

map和flatmap

除了map和flatMap之外，还有其他操作符以供使用。其他常用的操作符如下：

• filter：集合进行过滤
• each：遍历集合
• take：取出集合中的前几个
• skip：跳过前几个元素

map

在使用map时涉及到一个接口，Func1。
Func1与Action1很相似，区别在于Func1包装的有返回值方法。
得到多个Person对象中的name，保存到nameList中

Observable.just(person1, person2, person3)
                //使用map进行转换，参数1：转换前的类型，参数2：转换后的类型
                .map(new Func1<Person, String>() {
                    @Override
                    public String call(Person p) {
                        String name = p.getName();//获取Person对象中的name
                        return name;//返回name
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        nameList.add(s);
                    }
                });

多次使用map，想用几个用几个

        Observable.just("Hello", "RxJava")
                .map(new Func1<String, Integer>() {//将String类型的转化为Integer类型的哈希码
                    @Override
                    public Integer call(String s) {
                        return s.hashCode();
                    }
                })
                .map(new Func1<Integer, String>() {//将转化后得到的Integer类型的哈希码再转化为String类型
                    @Override
                    public String call(Integer integer) {
                        return integer.intValue() + "";
                    }
                })
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, s);
                    }
                });

flatmap

flatmap和map的不同在于，flatmap进行遍历操作时不需要用循环语句。
使用map来实现打印所有学生所修个课程名

List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Action1<List<Course>> action1 = new Action1<List<Course>>() {
            @Override
            public void call(List<Course> courses) {
                //遍历courses，输出cuouses的name
                 for (int i = 0; i < courses.size(); i++){
                    Log.i(TAG, courses.get(i).getName());
                }
            }
        };
        Observable.from(students)
                .map(new Func1<Student, List<Course>>() {
                    @Override
                    public List<Course> call(Student student) {
                        //返回coursesList
                        return student.getCoursesList();
                    }
                })
                .subscribe(action1);

在Action1中出现了for来循环打印课程名，使用RxJava就是为了剔除这样的嵌套结构。

List<Student> students = new ArrayList<Student>();
        students.add...
        ...

        Observable.from(students)
                .flatMap(new Func1<Student, Observable<Course>>() {
                    @Override
                    public Observable<Course> call(Student student) {
                        return Observable.from(student.getCoursesList());
                    }
                })
                .subscribe(new Action1<Course>() {
                    @Override
                    public void call(Course course) {
                        Log.i(TAG, course.getName());
                    }
                });

flatmap对数据做的合并操作，不能保证数据传入和取出的顺序一样，也就是说flatmap的操作是无序的；如果想要有序的效果，使用concatMap。

Scheduler

Scheduler：线程控制器，可以指定每一段代码在什么样的线程中执行。
我个人对观察者模式的理解就是为了解决，异步操作；后台处理逻辑，处理结果回调到前台显示。所以Scheduler线程控制对RxJava来说相当重要！

在RxJava中有几种不同的Scheduler:

• Schedulers.immediate()：直接在当前线程运行，相当于不指定线程。这是默认的 Scheduler。
• 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 主线程运行。

新的线程发起事件，在主线程中消费

Observable.just("Hello", "Word")
                .subscribeOn(Schedulers.newThread())//指定 subscribe() 发生在新的线程
                .observeOn(AndroidSchedulers.mainThread())// 指定 Subscriber 的回调发生在主线程
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        Log.i(TAG, s);
                    }
                });

subscribeOn()，和observeOn()方法来指定发生的线程和消费的线程。

• subscribeOn()：指定subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。
• observeOn()：指定Subscriber 所运行在的线程。或者叫做事件消费的线程。

以及参数Scheduler。
多次切换线程

Observable.just("Hello", "RxJava")
                .subscribeOn(Schedulers.newThread())//指定：在新的线程中发起
                .observeOn(Schedulers.io())         //指定：在io线程中处理
                .map(new Func1<String, String>() {
                    @Override
                    public String call(String s) {
                        return handleString(s);       //处理数据
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())//指定：在主线程中处理
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String s) {
                        show(s);                       //消费事件
                    }
                });

observeOn()被调用了两次，分别指定了map的处理的现场和消费事件show(s)的线程。

若将observeOn(AndroidSchedulers.mainThread())去掉会怎么样？不为消费事件show(s)指定线程后，map的处理和最后的消费事件show(s)都会在io线程中执行。observeOn() 指定的是它之后的操作所在的线程。
observeOn()可以多次使用，可以随意变换线程

RxJava与Retrofit结合使用

使用Retrofit传统方式定义接口和使用

//定义接口
@GET("/user")
public void getUser(@Query("userId") String userId, Callback<User> callback);
//使用
getUser(userId, new Callback<User>() {
    @Override
    public void success(User user) {
        userView.setUser(user);
    }

    @Override
    public void failure(RetrofitError error) {
        // Error handling
        ...
    }
};

使用Retrofit+Rxjava定义接口和使用

//定义接口
@GET("/user")
public Observable<User> getUser(@Query("userId") String userId);
//使用
getUser(userId)
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<User>() {
        @Override
        public void onNext(User user) {
            userView.setUser(user);
        }

        @Override
        public void onCompleted() {
        }

        @Override
        public void onError(Throwable error) {
            // Error handling
            ...
        }
    });

Retrofit 把请求封装进 Observable ，在请求结束后调用 onNext() 或在请求失败后调用 onError()。

使用普通方式修改数据库中User的值，耗时操作放在工作线程执行，是异步操作。

getUser(userId, new Callback<User>() {
    @Override
    public void success(User user) {
        new Thread() {
            @Override
            public void run() {
                processUser(user); // 尝试修正 User 数据
                runOnUiThread(new Runnable() { // 切回 UI 线程
                    @Override
                    public void run() {
                        userView.setUser(user);
                    }
                });
            }).start();
    }

    @Override
    public void failure(RetrofitError error) {
        // Error handling
        ...
    }
};

使用RxJava

getUser(userId)
    .doOnNext(new Action1<User>() {
        @Override
        public void call(User user) {
            processUser(user);
        })
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<User>() {
        @Override
        public void onNext(User user) {
            userView.setUser(user);
        }

        @Override
        public void onCompleted() {
        }

        @Override
        public void onError(Throwable error) {
            // Error handling
            ...
        }
    });