RxJava介绍

RxJava github地址 https://github.com/ReactiveX/RxJava
ReactiveX 官网 http://reactivex.io/intro.html

RxJava 是什么

RxJava is a Java VM implementation of Reactive Extensions: a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent
data structures.

rxjava是用java实现Reactive Extensions，可观测的序列来组成异步的、基于事件的程序的库。它扩展了观察者模式来支持数据/事件序列，而抽象掉的事情，如低级别的线程，同步，线程安全和并发数据结构。（翻译的有些绕，简而言之，就是做异步处理）

RxJava是一种观察者模式。

RxJava 的优势

AsyncTask ， HandlerThread 等都可可以做异步处理，RxJava有什么优势呢？一个词：简洁。而且RxJava能够在程序逻辑变得很复杂的情况下，依然保持简洁。

RxJava的使用方式

RxJava 的基本实现主要有三点：

（1）创建 Observer

Observer 即观察者，它决定事件触发的时候将有怎样的行为。 RxJava 中的 Observer 接口的实现方式，除了 Observer 接口之外，RxJava 还内置了一个实现了 Observer 的抽象类：Subscriber。 Subscriber 对 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!");
}
};

（2）创建 Observable

Observable 即被观察者，它决定什么时候触发事件以及触发怎样的事件。 RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则：

//可观察者
Observable observable = Observable.create(new Observable.OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
Log.d(tag, "call: ");
subscriber.onNext("next");
subscriber.onCompleted();
}
});

（3）Subscribe (订阅)

创建了 Observable 和 Observer 之后，再用 subscribe() 方法将它们联结起来，整条链子就可以工作了。需要特别注意的是，如果不执行subscribe，Observer与Observable都不会执行。这与普通的观察这模式不同。

observable.subscribe(observer);

（4）综合使用

Subscriber<? super Object> subscriber = new Subscriber<Object>() {

@Override
public void onCompleted() {
Log.d(tag, "subscriber: onCompleted ");
}

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

@Override
public void onNext(Object o) {
Log.d(tag, "subscriber: onNext " + o);
}
};
//  通过 observeOn() 的多次调用，程序实现了线程的多次切换。
//   不过，不同于 observeOn ()，subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。
Func1<? super Object, ?> mapOperator = new Func1<Object, Object>() {
@Override
public Object call(Object o) {
Log.d(tag, "mapOperator: call " + o.toString());
return " call1 " + o.toString();
}
};
Func1<? super Object, ?> mapOperator2 = new Func1<Object, Object>() {
@Override
public Object call(Object o) {
Log.d(tag, "mapOperator2: call " + o.toString());
return " call2 " + o.toString();
}
};
Observable.just("1", "2", "3", "4") // IO 线程，由 subscribeOn() 指定
.flatMap(new Func1<String, Observable<?>>() {
@Override
public Observable<?> call(String s) {
return Observable.just("number" + s);
}
})
.subscribeOn(Schedulers.io())            //subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。
.observeOn(Schedulers.newThread())
.map(mapOperator) // 新线程，由 observeOn() 指定
.observeOn(Schedulers.io())
.map(mapOperator2) // IO 线程，由 observeOn() 指定
.observeOn(AndroidSchedulers.mainThread())
.subscribe(subscriber);  // Android 主线程，由 observeOn() 指定

05-25 17:46:21.639 19868-19982/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator: call number1
05-25 17:46:21.640 19868-19981/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator2: call  call1 number1
05-25 17:46:21.642 19868-19982/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator: call number2
05-25 17:46:21.642 19868-19981/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator2: call  call1 number2
05-25 17:46:21.643 19868-19982/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator: call number3
05-25 17:46:21.643 19868-19981/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator2: call  call1 number3
05-25 17:46:21.644 19868-19982/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator: call number4
05-25 17:46:21.644 19868-19981/com.example.jiafei.testretrofit2 D/testrxjava: mapOperator2: call  call1 number4
05-25 17:46:21.652 19868-19868/com.example.jiafei.testretrofit2 D/testrxjava: subscriber: onNext  call2  call1 number1
05-25 17:46:21.652 19868-19868/com.example.jiafei.testretrofit2 D/testrxjava: subscriber: onNext  call2  call1 number2
05-25 17:46:21.652 19868-19868/com.example.jiafei.testretrofit2 D/testrxjava: subscriber: onNext  call2  call1 number3
05-25 17:46:21.652 19868-19868/com.example.jiafei.testretrofit2 D/testrxjava: subscriber: onNext  call2  call1 number4
05-25 17:46:21.652 19868-19868/com.example.jiafei.testretrofit2 D/testrxjava: subscriber: onCompleted

Scheduler 的 API

在RxJava 中，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。

另外， Android 还有一个专用的 AndroidSchedulers.mainThread()，它指定的操作将在 Android 主线程运行。

有了这几个 Scheduler ，就可以使用 subscribeOn() 和 observeOn() 两个方法来对线程进行控制了。 * subscribeOn(): 指定 subscribe() 所发生的线程，即 Observable.OnSubscribe 被激活时所处的线程。或者叫做事件产生的线程。 * observeOn(): 指定 Subscriber 所运行在的线程。或者叫做事件消费的线程。，通过 observeOn() 的多次调用，程序实现了线程的多次切换。不过，不同于
observeOn() ， subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。

欢迎扫描二维码，关注公共账号