前言
随着RxJava越来越火爆，很多人想入门却不知道从何做起，笔者整理了一天，总结出一些比较实用的东西，望各位看官斧正。
一、RxJava到底是什么，有什么好处？
关键词：异步、简洁。
RxJava可以进行异步调用，使每一步都可以自行处理其所在的线程，典型的流式代码结构，像读一首诗一样顺畅。即使以后逻辑越来越复杂，关于Rx的代码依然很顺畅。为什么这么说，请往下看。
二、RxJava例子
先比较一下下方的两个代码：
new Thread(){
@Override
public void run() {
super.run();
//网络请求~~~~~之后
{
runOnUiThread(new Runnable() {
@Override
public void run() {
ts.setText("这是通过newThread");
}
});
}
}
}.start();
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
//异步操作网络请求后
{
subscriber.onNext("这是通过RxJava");
subscriber.onCompleted();
}
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<Object>() {
@Override
public void call(Object o) {
ts2.setText(o.toString());
}
});
同样是开启新线程，从异步线程获取数据，然后再UI线程中写入，代码量相比较，下方的要比上面的多，但是对于易读性来讲，流式的代码结构，从上向下阅读，条理清晰，操作简单，不论以后逻辑修改多么复杂，几乎不会有太大的变化，不像第一部分的代码，如果说复杂度增加的话，就会很麻烦。
三、RxJava使用方法
本文是基于Intelli JDEA 进行的开发，使用了gradle，只需要加一行配置引入即可：
compile 'io.reactivex:rxjava:1.2.1'
概念基础
RxJava使用的设计模式为观察者模式，主要的步骤就是创建观察者，被观察者，订阅。比较熟悉的同学可以略过这一段。
读书人能叫盗么，这叫借.png
如上图所示，观察者模式可有四个模块：
被观察者接口类：主要提供接口声明包括订阅 解除订阅 通知功能，和保存观察者对象。
观察者接口类：主要提供一个方法供被观察者使用，当被观察者触发了一些条件后，通知观察者。
被观察者实现类及观察者实现类：实现以上接口。
举个简单的例子，在家懒得做饭订了外卖，跟送外卖的说（被观察者），等你到楼下了给我打电话（订阅），等外卖到的时候，就会给你打电话，通知你去取外卖（通知），这就是一个观察者模式应用。
使用流程
RxJava的开发主要分为三个步骤:apple、pen、Ah~
皮.jpg
1、Observer (观察者)
Observer属于观察者，我们可以看一下源码：
public interface Observer<T> {
onCompleted();
onError(Throwable e);
void onNext(T t);
}
很简单，就三个方法，代表着完成（onCompleted），错误（onError），接收数据（onNext）。
另外还有一个实现了Observer的抽象类Subscriber，主要方法和Observer差不多，比较大的区别就是可以使用onStart()方法和unsubscribe()方法。
onStart()：主要在事件未发出的时候调用，仅能在subscribe 所发生的线程被调用，可以做一些数据处理。
unsubscribe()：用于取消订阅。在这个方法被调用后，Subscriber 将不再接收事件。主要用于当某些对象不再使用的时候，但是并没有收到onCompleted消息的时候调用，防止出现内存泄露。
主要实现代码：
Subscriber observer = new Subscriber() {
@Override
public void onCompleted() {
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Object o) {
}
@Override
public void onStart() {
super.onStart();
System.out.println("----onStart----");
}
};
2、Observable(被观察者)
Observable为被观察者，由被观察者来决定是否发送消息，何时发送消息，异步操作一般都在这里进行，实现方式：
Observable observable = Observable.create(new Observable.OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
subscriber.onNext("111");
subscriber.onCompleted();
}
});
调用Observable的create方法来创建被观察者，其中call方法的参数即可视为第一步的Observer ，在进行订阅的时候，会直接执行call方法内的代码。
除了create方法外，还有简单的方法来直接实现，just(T...)方法和from(T[]) 方法
String[] txt = {"easd","dasd","dasd"};
Observable observable = Observable.from(txt);
Observable observable2 = Observable.just("easd","dasd","dasd");
查看源码后，其实实现的方法也是create方法，只是经过了一些简单的封装，这里就简单介绍一下from方法。
public static <T> Observable<T> from(T[] array) {
int n = array.length;
if (n == 0) {
return empty();
} else
if (n == 1) {
return just(array[0]);
}
return create(new OnSubscribeFromArray<T>(array));
}
又上面代码可知，from方法首先对数组进行了判断，然后return了一个Observable的对象，使用的是create()方法，那么OnSubscribeFromArray又做了什么呢?
public final class OnSubscribeFromArray<T> implements OnSubscribe<T>
这是OnSubscribeFromArray实际上也是实现了OnSubscribe接口，那么我们去看一下call()方法：
@Override
public void request(long n) {
if (n < 0) {
throw new IllegalArgumentException("n >= 0 required but it was " + n);
}
if (n == Long.MAX_VALUE) {
if (BackpressureUtils.getAndAddRequest(this, n) == 0) {
fastPath();
}
} else
if (n != 0) {
if (BackpressureUtils.getAndAddRequest(this, n) == 0) {
slowPath(n);
}
}
}
void fastPath() {
final Subscriber<? super T> child = this.child;
for (T t : array) {
if (child.isUnsubscribed()) {
return;
}
child.onNext(t);
}
if (child.isUnsubscribed()) {
return;
}
child.onCompleted();
}
void slowPath(long r) {
final Subscriber<? super T> child = this.child;
final T[] array = this.array;
final int n = array.length;
long e = 0L;
int i = index;
for (;;) {
while (r != 0L && i != n) {
if (child.isUnsubscribed()) {
return;
}
child.onNext(array[i]);
i++;
if (i == n) {
if (!child.isUnsubscribed()) {
child.onCompleted();
}
return;
}
r--;
e--;
}
r = get() + e;
if (r == 0L) {
index = i;
r = addAndGet(e);
if (r == 0L) {
return;
}
e = 0L;
}
}
}
主要调用了这三个方法，当数组长度等于long的最大值时，调用fastPath()，反之则 slowPath(n)，两个方法的主要目的其实也就是将数组进行遍历，然后分发出去。由此可知无论是just()还是from()都可以视为与create()方法等价。
3、Subscribe （订阅）
最激动人心的时候到了，之前的准备已经完成了之后，就可以将观察者和被观察者进行订阅。
observable.subscribe(observer);
observable.subscribe(subscriber);
Ah~~~~这样关系就定下来了。subscribe方法代码比较啰嗦，就简化一下发到下面：
static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
if (subscriber == null) {
throw new IllegalArgumentException("subscriber can not be null");
}
if (observable.onSubscribe == null) {
throw new IllegalStateException("onSubscribe function can not be null.");
}
//1 !!
subscriber.onStart();
try {
//2 !!
RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);
return RxJavaHooks.onObservableReturn(subscriber);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
if (subscriber.isUnsubscribed()) {
RxJavaHooks.onError(RxJavaHooks.onObservableError(e));
} else {
try {
subscriber.onError(RxJavaHooks.onObservableError(e));
} catch (Throwable e2) {
Exceptions.throwIfFatal(e2);
RuntimeException r = new OnErrorFailedException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
RxJavaHooks.onObservableError(r);
throw r;
}
}
return Subscriptions.unsubscribed();
}
}
主要就三步：
一调用subscriber的onStart()方法，
二调用被观察者observable的OnSubscribe的call方法，
三将subscriber变成Subscription 返回，可以直接使用unsubscribed()方法。
另外，subscribe方法支持不完整的回调，可以使用Action1和Action0作为回调：
Action1<String> onNextAction = new Action1<String>() {
// onNext()
@Override
public void call(String s) {
Log.d(tag, s);
}
};
Action1<Throwable> onErrorAction = new Action1<Throwable>() {
// onError()
@Override
public void call(Throwable throwable) {
// Error handling
}
};
Action0 onCompletedAction = new Action0() {
// onCompleted()
@Override
public void call() {
Log.d(tag, "completed");
}
};
observable.subscribe(onNextAction，onErrorAction，onCompletedAction);
//分别对应onNext，onError，onComplete。
//也可以
observable.subscribe(onNextAction，onErrorAction);
observable.subscribe(onNextAction);
//在源码里都有不同的方法去对应
以上所说都是关于RxJava的使用方法，但是！跟特么所说的异步没毛关系，并没有什么卵用，下面开始正片
三、RxJava的异步调度
这个东西专门拿出来当一个大分类来讲，在RxJava里如果没有这个东西，要他有何用。这个神奇的东西叫Scheduler，线程调度器。主要用在两个方法里，observeOn()和subscribeOn(),分别对应观察者操作线程和被观察操作线程。
总共来讲有这么几个线程：
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 主线程运行。
还有个Schedulers.from(Executor executor),你可以自定义线程池来处理。
是不是很吊！
再次回到一开始放出来的代码：
Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
//异步操作网络请求后
{
subscriber.onNext("这是通过RxJava");
subscriber.onCompleted();
}
}
})
.subscribeOn(Schedulers.io())//Observable操作放到io线程里
.observeOn(AndroidSchedulers.mainThread())//Observable操作放在UI线程里
.subscribe(new Action1<Object>() {//缺省的onNext调用
@Override
public void call(Object o) {
ts2.setText(o.toString());
}
});
这样是不是就很容易理解了？
今天就先到此为止，下一节主要介绍变换和Subject这个神奇的东西
网友评论