前言
使用了RxJava有一段时间了，深深感受到了其“牛逼”之处。下面，就从RxJava的基础开始，一步一步与大家分享一下这个强大的异步库的用法！
Github Repo上给的解释是：“RxJava is a Java VM implementation of Reactive Extensions:
a library for composing asynchronous and event-based programs by using observable sequences.”
大概就是说RxJava是Java VM上一个灵活的、使用可观测序列来组成的一个异步的、基于事件的库。咋一看好像不知道是啥东西… … 没事，往下看~
上面 这段解释，重点就在于异步！但是它又不像 AsyncTask 这样用法简单，所以刚接触RxJava的童鞋，可能会觉得特别难，无从下手，没事，相信通过这篇文章，大伙儿可以有一个比较深刻的理解！
RxJava精华可以浓缩为异步两个字，其核心的东西不外乎两个：
1. Observable（被观察者） 2. Observer/Subscriber（观察者）
Observables可以发出一系列的 事件，这里的事件可以是任何东西，例如网络请求、复杂计算处理、数据库操作、文件操作等等，事件执行结束后交给 Observer/Subscriber 的回调处理。
Maven/RxJava下获取最新版本。
RxJava常用操作符后面会详细介绍，这里先有个了解。// 创建对象,just里面的每一个参数，相当于调用一次Subscriber#OnNext()Observable observable = Observable.just("Hello World!");
// 创建对象,just里面的每一个参数，相当于调用一次Subscriber#OnNext()Observableobservable = Observable.just("Hello World!");
这样，是不是简单了许多？
上述事件进行消费。首先，先创建一个观察者。// 创建一个ObserverObserver observer = new Observer() { @Override
public void onCompleted() {
Log.i(TAG, "complete");
} @Override
public void onError(Throwable e) {
} @Override
public void onNext(String s) {
Log.i(TAG, s);
}
};
或者
// 创建一个SubscriberSubscribersubscriber = new Subscriber () { @Override public void onCompleted() { Log.i(TAG, "complete"); } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { Log.i(TAG, s); } };
Observer 是观察者， Subscriber 也是观察者，Subscriber 是一个实现了Observer接口的抽象类，对 Observer 进行了部分扩展，在使用上基本没有区别；
Subscriber 多了发送之前调用的
onStart()和解除订阅关系的unsubscribe()方法。并且，在 RxJava 的 subscribe 过程中，Observer 也总是会先被转换成一个 Subscriber 再使用。所以在这之后的示例代码，都使用 Subscriber 来作为观察者。
事件消费与事件订阅来说，好像为了打印一个“Hello World！”要费好大的劲… 其实，RxJava 自身提供了精简回调方式，我们可以为 Subscriber 中的三种状态根据自身需要分别创建一个回调动作 Action：// onComplete()Action0 onCompleteAction = new Action0() { @Override
public void call() {
Log.i(TAG, "complete");
}
};// onNext(T t)Action1 onNextAction = new Action1() { @Override
public void call(String s) {
Log.i(TAG, s);
}
};// onError(Throwable t)Action1 onErrorAction = new Action1() { @Override
public void call(Throwable throwable) {
}
};
那么，RxJava 的事件订阅支持以下三种不完整定义的回调。
observable.subscribe(onNextAction); observable.subscribe(onNextAction, onErrorAction); observable.subscribe(onNextAction, onErrorAction, onCompleteAction);
我们可以根据当前需要，传入对应的 Action， RxJava 会相应的自动创建 Subscriber。
Action0 表示一个无回调参数的Action；
Action1 表示一个含有一个回调参数的Action；
当然，还有Action2 ~ Action9，分别对应2~9个参数的Action；
每个Action，都有一个 call() 方法，通过泛型T，来指定对应参数的类型；
上述示例 进行改写。final ImageView ivLogo = (ImageView) findViewById(R.id.ivLogo);
Observable.create(new Observable.OnSubscribe() { @Override
public void call(Subscriber super String> subscriber) {
subscriber.onNext("https://ss2.baidu.com/-vo3dSag_xI4khGko9WTAnF6hhy/image/h%3D200/sign=4db5130a073b5bb5a1d727fe06d2d523/cf1b9d16fdfaaf51965f931e885494eef11f7ad6.jpg");
}
}).map(new Func1() { @Override
public Drawable call(String url) { try {
Drawable drawable = Drawable.createFromStream(new URL(url).openStream(), "src"); return drawable;
} catch (IOException e) {
} return null;
}
}) // 指定 subscribe() 所在的线程，也就是call()方法调用的线程
.subscribeOn(Schedulers.io()) // 指定 Subscriber 回调方法所在的线程，也就是onCompleted, onError, onNext回调的线程
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber() { @Override
public void onCompleted() {
} @Override
public void onError(Throwable e) {
Log.e(TAG, e.toString());
} @Override
public void onNext(Drawable drawable) { if (drawable != null) {
ivLogo.setImageDrawable(drawable);
}
}
});
经过改写代码后，有什么变化呢？ Observable 创建了一个 String 事件，也就是产生一个url，通过 map 操作符进行变换，返回Drawable对象，这个变换指的就是通过url进行网络图片请求，返回一个Drawable。所以简单的来说就是把String事件，转换为Drawable事件。逻辑表示就是：
Observable--> map变换 --> Observable
那么，Func1 是什么呢？与 Action1 类似,不同的是 FuncX 有返回值，而 ActionX 没有。为什么需要返回值呢？目的就在于对象的变换，由String对象转换为Drawable对象。同样，也有Func0 ~ Func9，对应不同的参数个数。
当然了，RxJava 的变换，可不止于map这么简单，继续往下！
15.Subject会介绍这个。为了演示效果，先举个简单栗子：Observable
.create(new Observable.OnSubscribe() { @Override
public void call(Subscriber super Integer> subscriber) { int i = 0; int[] times = new int[]{100, 1000}; while (true) {
i++; if (i >= 100) break;
subscriber.onNext(i); try { // 注意！！！！
// 当i为奇数时，休眠1000ms，然后才发送i+1，这时i不会被过滤掉
// 当i为偶数时，只休眠100ms，便发送i+1，这时i会被过滤掉
Thread.sleep(times[i % 2]);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
subscriber.onCompleted();
}
}) // 间隔400ms以内的事件将被丢弃
.debounce(400, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber() { @Override
public void onCompleted() {
Log.i(TAG, "complete");
} @Override
public void onError(Throwable e) {
Log.e(TAG, e.toString());
} @Override
public void onNext(Integer integer) {
Log.i(TAG, "integer = " + integer);
}
});
输出结果：
11-23 10:44:45.167 MainActivity: integer = 111-23 10:44:46.270 MainActivity: integer = 311-23 10:44:47.373 MainActivity: integer = 511-23 10:44:48.470 MainActivity: integer = 711-23 10:44:49.570 MainActivity: integer = 911-23 10:44:50.671 MainActivity: integer = 1111-23 10:44:51.772 MainActivity: integer = 1311-23 10:44:52.872 MainActivity: integer = 1511-23 10:44:53.973 MainActivity: integer = 17...
我们设置过滤条件为400ms，可以发现，奇数正常输出，因为在它的下一个事件事件隔了1000ms，所以它不会被过滤掉；偶数被过滤掉，是因为它距离下一个事件（奇数）只隔了100ms。并且，输出的两个事件相隔大约为 100ms + 1000ms = 1100ms。
merge
用于合并两个Observable为一个Observable。较为简单。
Observable.merge(Observable1, Observable2) .subscribe(subscriber);
concat
顺序执行多个Observable，个数为1 ~ 9。例子稍后与first操作符一起~~
compose
与 flatMap 类似，都是进行变换，返回Observable对象，激活并发送事件。
compose是唯一一个能够从数据流中得到原始Observable的操作符，所以，那些需要对整个数据流产生作用的操作（比如，subscribeOn()和observeOn()）需要使用compose来实现。相较而言，如果在flatMap()中使用subscribeOn()或者observeOn()，那么它仅仅对在flatMap中创建的Observable起作用，而不会对剩下的流产生影响。这样就可以简化subscribeOn()以及observeOn()的调用次数了。compose是对 Observable 整体的变换，换句话说，flatMap转换Observable里的每一个事件，而compose转换的是整个Observable数据流。flatMap每发送一个事件都创建一个 Observable，所以效率较低。而compose操作符只在主干数据流上执行操作。建议使用
compose代替flatMap。
first
只发送符合条件的第一个事件。可以与前面的contact操作符，做网络缓存。举个栗子：依次检查Disk与Network，如果Disk存在缓存，则不做网络请求，否则进行网络请求。
// 从缓存获取ObservablefromDisk = Observable.create(new Observable.OnSubscribe () { @Override public void call(Subscriber super BookList> subscriber) { BookList list = getFromDisk(); if (list != null) { subscriber.onNext(list); } else { subscriber.onCompleted(); } } });// 从网络获取Observable fromNetWork = bookApi.getBookDetailDisscussionList(); Observable.concat(fromDisk, fromNetWork) // 如果缓存不为null，则不再进行网络请求。反之 .first() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber () { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(BookList discussionList) { } });
网络缓存用法，具体可参见我的项目：https://github.com/JustWayward/BookReader
timer
可以做定时操作，换句话讲，就是延迟执行。事件间隔由timer控制。举个栗子：两秒后输出“Hello World！”
Observable.timer(2, TimeUnit.SECONDS) .subscribe(new Subscriber() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Long aLong) { Log.i(TAG, "Hello World!"); } });
interval
定时的周期性操作，与timer的区别就在于它可以重复操作。事件间隔由interval控制。举个栗子：每隔两秒输出“Hello World！”
Observable.interval(2, TimeUnit.SECONDS) .subscribe(new Subscriber() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(Long aLong) { Log.i(TAG, "Hello World!"); } });
throttleFirst
与debounce类似，也是时间间隔太短，就丢弃事件。可以用于防抖操作，比如防止双击。
RxView.clicks(button) .throttleFirst(1, TimeUnit.SECONDS) .subscribe(new Observer
上面这个RxView详见：https://github.com/JakeWharton/RxBinding， 主要与RxJava结合用于一些View的事件绑定，JakeWharton大神的项目，厉害。
Single
Single与Observable类似，相当于是他的精简版。订阅者回调的不是OnNext/OnError/onCompleted，而是回调OnSuccess/OnError。
Single.create(new Single.OnSubscribe
Subject
Subject这个类，既是Observable又是Observer，啥意思呢？就是它自身既是事件的生产者，又是事件的消费者，相当于自身是一条管道，从一端进，又从另一端出。举个栗子：PublishSubject
Subject subject = PublishSubject.create();// 1.由于Subject是Observable，所以进行订阅subject.subscribe(new Subscriber
这个好像有点厉害的样子，哈哈。可以配合debounce，避免SearchEditText频繁请求。
Subject subject = PublishSubject.create(); subject.debounce(400, TimeUnit.MILLISECONDS) .subscribe(new Subscriber
flatMap操作符最后的获取Readme内容显示在WebView上的例子。Retrofit2 + RxJava + Dagger2: 具体可参见我的项目，里面有比较详细的用法。
https://github.com/JustWayward/BookReader
Retrofit2 + RxJava + Dagger2: 具体可参见我的项目，里面有比较详细的用法。
https://github.com/JustWayward/BookReader
不难发现，Retrofit 把请求封装进 Observable ，在请求结束后调用 onNext() 以及 OnCompleted() 或在请求失败后调用 onError()。
注：RxJava形式的请求，并不能减少代码量，但是逻辑非常清晰。假如请求到数据之后需要对数据进行处理，并且是耗时操作，难道要再开一个线程，或者用AsyncTask再做一次异步？很显然，RxJava的变换很好的解决了这个问题，依然会使逻辑结构清晰。
https://github.com/AndroidKnife/RxBus这个库。
https://github.com/JakeWharton/RxBinding， 主要与RxJava结合用于一些View的事件绑定。
12.interval做周期性操作的例子，并没有使之停下来的，没有去控制订阅的生命周期，这样，就有可能引发内存泄漏。所以，在Activity#onDestroy()的时候或者不需要继续执行的时候应该取消订阅。Subscription subscription = Observable.interval(2, TimeUnit.SECONDS)
.subscribe(new Subscriber() { @Override
public void onCompleted() {
} @Override
public void onError(Throwable e) {
} @Override
public void onNext(Long aLong) {
Log.i(TAG, "Hello World!");
}
});// 调用unsubscribe();方法进行取消订阅subscription.unsubscribe();
12.interval做周期性操作的例子，并没有使之停下来的，没有去控制订阅的生命周期，这样，就有可能引发内存泄漏。所以，在Activity#onDestroy()的时候或者不需要继续执行的时候应该取消订阅。Subscription subscription = Observable.interval(2, TimeUnit.SECONDS)
.subscribe(new Subscriber() { @Override
public void onCompleted() {
} @Override
public void onError(Throwable e) {
} @Override
public void onNext(Long aLong) {
Log.i(TAG, "Hello World!");
}
});// 调用unsubscribe();方法进行取消订阅subscription.unsubscribe();
但是，如果有很多个数据源，那岂不是要取消很多次？当然不是的，可以利用 CompositeSubscription， 相当于一个 Subscription 集合。
CompositeSubscription list = new CompositeSubscription(); list.add(subscription1); list.add(subscription2); list.add(subscription3);// 统一调用一次unsubscribe，就可以把所有的订阅都取消list.unsubscribe();
http://www.apkbus.com/blog-873055-77431.html
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/4822/viewspace-2811953/ ，如需转载，请注明出处，否则将追究法律责任。
