一、RxJava简介
- RxJava官方的解释如下：
要想了解RxJava首先需要了解Rx，Rx的全称是reactive extension，翻译一下就是响应式扩展，Rx是基于观察者模式的一种编程模型，目标是提供一致的编程接口，帮助开发者更方便地处理异步数据流，Rx渗透到了各种语言中，RxJava就是针对Java语言的一个异步的响应式编程库，它是对观察者模式的扩展。很多Android开发者都非常喜欢这个简洁的编程库。
- RxJava源码：
https://github.com/ReactiveX/RxJava/tree/2.x/src
二、观察者模式
- 观察者模式概念
观察者模式是将观察者与被观察者分离开，实现了对象间一种一对多的组合关系，当被观察者的状态发生变化时，所有依赖于它的观察者就会检测到变化并且刷新自己的状态。
- 观察者模式中的四个重要角色
抽象主题：定义添加和删除观察者的功能；
抽象观察者：定义观察者收到主题通知后要做什么事情；
具体主题：抽象主题的具体实现；
具体观察者：抽象观察者的具体实现。
大概的步骤就是1、创建被观察者；2、创建观察者；3、为被观察者添加观察者；4、被观察者通知所有的观察者发生变化。了解了观察者模式的大致实现步骤帮助我们更好的理解RxJava中的观察者模式的实现。
三、从RxJava源码中看观察者模式
依然从上文中提到的四个角色结合实现的四个步骤来看RxJava源码中的实现：
- 抽象主题：
public abstract class Observable<T> implements ObservableSource<T> {
……
}
这是一个Observable类，实现ObservableSource的接口；
public interface ObservableSource<T> {
/**
* Subscribes the given Observer to this ObservableSource instance.
* @param observer the Observer, not null
* @throws NullPointerException if {@code observer} is null
*/
void subscribe(@NonNull Observer<? super T> observer);
}
ObservableSource接口中subscribe用来实现订阅观察者角色的功能，这里的ObservableSource就是抽象主题的角色。
- 抽象观察者
public interface Observer<T> {
void onSubscribe(@NonNull Disposable d);
void onNext(@NonNull T t);
void onError(@NonNull Throwable e);
void onComplete();
}
Observable就是观察者的角色，上面就是Observer接口，定义了观察者收到被观察者的通知后要做的事情。
- 具体主题：
有了抽象主题，下一步就是实现观察者模式的第一步，创建具体主题也就是被观察者；
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
RxJava提供了很多的创建的方法，这是其中一种create；
Observable observable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello");
subscriber.onNext("Hi");
subscriber.onNext("Aloha");
subscriber.onCompleted();
}
});
类似于以上的方式，创建了Observable并且定义了一些事件触发的规则，create时，传入了一个OnSubscribe参数，它相当于一个计划表，当该Observable被订阅时，它就会调用call（）方法，触发定义好的事件。
- 具体观察者
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!");
}
};
RxJava中的observer接口的实现方式如上。
- 订阅
在观察者和被观察者都被创建之后，最重要的就是将两者结合起来，也就是实现订阅，订阅部分的代码非常简单，而且在我初次看到它的实现方式的时候觉得十分奇怪，从上文的创建Observable类中也可以看到，subscribe（）是被定义在Observable中的，这有点不符合我们平常的思考方式，据说这样更加符合流式API的设计。
observable.subscribe(observer);
还有一种写法是DefaultObserver，这里的DefaultObserver和observer的使用是相同的，subscriber是observer的抽象类。
下面从DefaultObserver的部分源码中看订阅的实现机制：
public abstract class DefaultSubscriber<T> implements FlowableSubscriber<T> {
Subscription upstream;
@Override
public final void onSubscribe(Subscription s) {
if (EndConsumerHelper.validate(this.upstream, s, getClass())) {
this.upstream = s;
onStart();
}
}
/**
* Requests from the upstream Subscription.
* @param n the request amount, positive
*/
protected final void request(long n) {
Subscription s = this.upstream;
if (s != null) {
s.request(n);
}
}
/**
* Cancels the upstream's Subscription.
*/
protected final void cancel() {
Subscription s = this.upstream;
this.upstream = SubscriptionHelper.CANCELLED;
s.cancel();
}
/**
* Called once the subscription has been set on this observer; override this
* to perform initialization or issue an initial request.
* <p>
* The default implementation requests {@link Long#MAX_VALUE}.
*/
protected void onStart() {
request(Long.MAX_VALUE);
}
}
可以看出DefaultObserver有四个方法：1、onStart（），这就是一个准备的方法；2、onSubscribe.call(subscriber)，这里事件发送的逻辑开始实现，因此可以看到Observable发送事件不是在创建的时候开始的，而是在建立了订阅这个连接的时候实现的；3、cancle（），取消一个订阅连接；4、request（），来自上游订阅的请求。（RxJava的源码注释中用上游表示主题，用下游表示观察者）
四、观察者模式的优点
观察者模式最主要的优点是它是低耦合的，也就是观察者和被观察者都只关注他们之间的接口，而不需要关心对方具体是哪个具体类的实现，当增加观察者和被观察者的时候，都对对方的具体实现没有影响。
观察者模式实现的是广播，被观察者可以向所有订阅了它的观察者发送事件。
一开始就说RxJava是对观察者模式的扩展，它对普通的观察者模式做出了一些调整，主要有：1、观察者通过onSubscribe（）获取了发送事件的Disposable对象，这样它就可以获取订阅连接中两者的状态，甚至可以主动的中断事件的发送，在普通的观察者模式中只有被观察者拥有订阅的集合并且控制它的订阅者；2、抽象主题并没有直接控制onNext（），onError（）这些事件，而是关注Emitter实例的发送，而具体的事件发送是在ObservableOnsubscribe接口监听到ObservableEmitter对象并且接受它之后才实现的。