（本文基于rxjava2.0/rxandroid2.0而写）
RxJava很神奇！很神奇不是指它在GitHub上写的“A library for composing asynchronous and event-based programs by using observable sequences.”，而是它的设计模型很神奇。
我解析了它的源码，发现RxJava竟然是类似干细胞进化。
RxJava的干细胞模型
RxJava的干细胞模型很简单，本质上是基于观察者模式的扩展，先上类图。
主类示意图
一个被观察者虚类Observable，实现它必须实现subscribeActual，一个观察者接口类Observer，它扩展了一些接口。OK，我们看看这个简单的结构，能干点什么。
首先，实现一个简单的Observable和Observer：
public class AObservable extends Observable<Integer>{
@Override
protected void subscribeActual(Observer<? super Integer> observer) {
int a = 1+2;
observer.onNext(a);
observer.onComplete();
}
}
(new AObservable()).subscribe(new Observer<Integer>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(Integer value) {
System.out.println(value);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
System.out.println("End!");
}
});
Observable做了两件事，计算1+2的结果，返回给Observer，Observer接收到结果后，打印了出来。在这里，Observable负责处理事情，Observer负责接收处理结果，就这么简单。这时候，你可能开始疑惑，就只能做这么一件事情？
当然不是！
Observable进化和相互作用
生物的干细胞能根据不同的需求，进化成肌细胞、血红细胞、脑细胞等等，Observable也不例外，通过继承，能实现不同的功能等Observable，而且RxJava库已经帮你实现了大量的Observable，你只需要简单地调用就行。
另外，进化后的细胞相互作用后，会形成各种各样的生理机能，Observable也是如此。想想，当BObservable给AObservable提供Observer的时候，会发生什么事？我们来看一个有趣的例子。
Observable.just(1, 2, 3,4 )
.map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "Hello, "+integer;
}
})
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
just(1, 2, 3,4 )实现了一个ObservableFromArray的Observable对象，map(...)实现了一个ObservableMap的Observable对象，并且在它的subscribeActual里面给ObservableFromArray提供了Observer。这样子，你就看到这样子的场景，ObservableFromArray把1234形成整型数组，挨个传给了ObservableMap，ObservableMap把整型数组挨个转成"Hello, x"后，传给它的Observer，打印出来。
你可以这样子随意地拼接能提供Observer的Observable，形成复杂的任务链。怎么样，是不是很神奇？RxJava可不是仅限于这种简单的拼接交互，不过更多的交互得你自己去挖掘了。
好了，我们来看一个更加神奇的进化。
进一步进化
RxJava有两个特殊的Observer，ObservableObserveOn和ObservableSubscribeOn，里面都包含了Scheduler，用于控制任务执行的线程。
下面，我们来看看源码，先看ObservableObserveOn。
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize)
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
ObservableObserveOn实现了一个观察者ObserveOnObserver，这个观察者关联了ObservableObserveOn的观察者，即subscribeActual传入参数Observer<? super T> observer，和Scheduler.Worker，Scheduler.Worker会把observer的监听函数放入到Scheduler创建的线程中运行。
再看ObservableSubscribeOn。
@Override
public void subscribeActual(final Observer<? super T> s) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);
s.onSubscribe(parent);
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
}
scheduler直接把 source.subscribe(parent);放到新的线程中执行，实现对source的subscribeActual线程控制，因为subcribe会执行subcribeActual。
更多的Observable或者Flowable（支持Backpressure的Observable），查看源码。
一个使用示例
Observable.just("one", "two", "three", "four", "five")
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
System.out.println(s);
}
});
这个示例创建了3个Observable，just("one", "two", "three", "four", "five")创建了ObservableFromArray，用于把数组中的子项一个个输出，subscribeOn(Schedulers.newThread())创建了ObservableSubscribeOn，用于控制ObservableFromArray的subscribe的执行线程，observeOn(AndroidSchedulers.mainThread())创建了ObservableObserveOn，用来控制接下来的Observe的监听线程。
看完，相信大家对 Rxjava的机制和用法都有了很直观的认识！
网友评论