什么是 rxJava?

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.
简单来说，rxJava 是一种 基于事件的，使用了可被观察序列 的异步 响应 扩展 的类库。

特性

• rxJava 是解决 异步问题的。
• rxJava 是基于事件机制的。
• rxJava 是处理响应扩展的，解决 事件的响应问题（比如事件的响应在哪个线程？）。
• rxJava 是序列的。它是个有顺序的队列，按顺序的，先进后出的。它对集合对象的支持很好。

rxJava 使用了 设计模式里的 观察者模式 来实现。它的核心理念的两个东西：

• 被观察者 被观察的对象，它是一个事件源，它的状态将会订阅者观察到。
• 观察者（订阅者） 关注“被观察者”的对象
• 订阅 建立关系，我们说“订阅者”订阅了“被观察者”

rxJava 可以用来改善用户操作体验，它很方便的切换代码运行的线程（UI线程或者工作线程），它与AsyncTask的功能类似，使得我们可以在工作线程共执行耗时的逻辑，完成后再UI线程处理视图状态的编号。
rxjava 能带来代码可读性，让代码清晰。

引用类库

compile 'io.reactivex:rxjava:1.0.14'
compile 'io.reactivex:rxandroid:1.0.1'

简单示例

Rxjava的代码很优雅，链式的写法，很清晰的表达了代码的行为。

Observable.from () 创建了一个被观察的对象。
　　.map() 方法执行了转换，将对象转成字符串
　　.subscribeOn() 指示了 在哪个线程里执行 上面的转换操作。
　　.observeOn() 指示了 在哪个线程里执行 对结果的响应操作，比如后面的 alert 方法。
　　.subscribe() 传递了具体的响应处理，即执行了 alert方法。

示例：

Observable.from(userBeans)
                .map(new Func1<UserBean, String>() {
                    @Override
                    public String call(UserBean userBean) {
                        //处理对象的转换
                        return String.format("%s(%s)", userBean.name, userBean.age);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<String>() {
                    @Override
                    public void call(String str) {
                       
                        alert(str); //在UI视图显示
                    }
                });

从单个对象创建被观察对象的演示

使用 just() 方法，指定单个对象来创建被观察对象实例。

private void doSomeOne() {
    Observable.just("Hello, world!").subscribe(new Action1<String>() {
        @Override
        public void call(String str) {
            alert(str);
        }
    });
}

从一个集合创建被观察对象的演示

使用 from() 方法，指定一个集合来创建被观察对象实例。

private void doSomeArray() {
    Observable.from(new String[]{"A", "B", "C"}).subscribe(new Action1<String>() {
        @Override
        public void call(String str) {
            alert(str);
        }
    });
}

对象转换的演示

我们使用 map() 方法进行转换。下面的代码演示了 从 UserBean数组创建被观察对象，然后使用map()方法转换成 字符串。

private void doSomeEntity() {
    UserBean[] userBeans = new UserBean[3];
    userBeans[0] = new UserBean("jo", 18);
    userBeans[1] = new UserBean("ken", 9);
    userBeans[2] = new UserBean("hack", 30);

    //增加了map方法，转出 实体到字符串
    Observable.from(userBeans).map(new Func1<UserBean, String>() {
        @Override
        public String call(UserBean userBean) {
            return String.format("%s(%s)", userBean.name, userBean.age);
        }
    }).subscribe(new Action1<String>() {
        @Override
        public void call(String str) {
            alert(str);
        }
    });
}

调度器 Scheduler

Scheduler 用以指定运行在哪个线程下。比如：

 .subscribeOn(Schedulers.io())                  指示运行在io线程
 .observeOn(AndroidSchedulers.mainThread())     指示运行在主线程

subscribeOn 方法

它指定 subscribe() 发生在 IO 线程. 比如：subscribeOn(Schedulers.io())
一般来说，我们常用的 处理图片，网络访问，会在这里用这个方法指定为 io 线程运行。

observeOn

它指定 Subscriber 的回调发生在主线程 .observeOn(AndroidSchedulers.mainThread())
一般来说，android下都会使用主线程以操作视图

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下特有的, 在 Android 主线程运行。

参考：

https://github.com/ReactiveX/RxJava
http://gank.io/post/560e15be2dca930e00da1083