不知不觉博客搭建完成已经一个多月了，至今磕磕绊绊才发了两篇。搭建自己的博客之前看过不少别人搭建的技术博客，隐约发现一个有意思的规律，似乎每个博客的开始段落都会放一张跟技术毫不相干的照片，比如风景照，明星，篮球，动漫等等。我看过的绝大多数技术博客都完美的符合这一规律，原因不详，不明觉厉。平常倒是见过很多餐厅里挂关公、宿舍里挂科比柯南的，不知道是不是同一个道理，emmm……そですね。 所以自己写博客的时候觉得也遵守一下规律，求以后写博客文思泉涌，周末不加班……。
版权声明：本文为博主原创文章，未经博主允许不得转载。
一起读RxJava源码系列：
一起读RxJava源码（一）——简介
一起读RxJava源码（二）——基础知识:观察者模式
一起读RxJava源码（三）——RxJava的基本实现
一起读RxJava源码（四）——转换操作符
一起读RxJava源码（五）——线程调度
一起读RxJava源码（六）——深入浅出：基于波浪事件流和模块化的思路分析RxJava
1. 前言
上一篇对RxJava做了个简单介绍，在一起读源码前还有一些基础知识要啰嗦两句，个人认为对理解RxJava源码思想有很大的帮助。
2. 观察者模式
RxJava的异步实现，是通过一种扩展的观察者模式来实现的。在之前，当然首先要理解观察者模式。
2.1 观察者模式定义
观察者模式定义了对象之间的一对多依赖，当一个对象改变状态时，它的所有依赖者都会收到通知并自动更新。
简单举个例子：
我们看看报纸和杂志的订阅是怎么回事：
1、 报社的业务就是出版报纸。
2、 向某家报社订阅报纸，只要他们有新报纸出版，就会给你送来。只要你是他们的订户，你就会一直收到新报纸。
3、 当你不想再看报纸的时候，取消订阅，他们就不会再送新报纸来。
4、 只要报社还在运营，就会一直有人（或单位）向他们订阅报纸或取消订阅报纸。
简单一句话概括就是：出版者+订阅者=观察者模式
报社和订阅者组成的模式其实就是观察者模式。翻译成普通话就是：主题(Subject)/被观察者(Observable) + 观察者(Observer) = 观察者模式。
2.2 观察者模式作用
重要作用：解耦。也就是将观察者和被观察者解耦。并且解决一对多依赖的问题。
2.3 观察者模式UML类图
2.4 java内置观察者模式源码解析
Java API提供了内置的观察者模式。Java.util包里包含最基本的Observer接口以及Observable类。下面我们来具体看一下他们的源码。
首先是java.util.Observable，这是一个被观察者的类，相当于出版者/社。这里只节选了主要方法和逻辑并进行了注释，方便理解原理。多说一句，其实类里的英文注释讲的非常清楚了，但是由于和中文注释混在一起实在太乱，于是这里把英文的删掉了。英文好的工友可以直接看类里的英文注释。1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87public class Observable {
private boolean changed = false;
private Vector<Observer> obs;
public Observable() {
obs = new Vector<>();
//这里初始化一个存放观察者“们”（Observers）的容器。可以理解为报社里存储订阅者地址的清单。它是线程安全的。
}
/**
* 这个方法是添加观察者的方法，被addObserver添加进来的不重复的Observer（观察者）放到Vector容器里，这个Observer（观察者）就和当前被观察者产生了依赖。
* 还是用订报纸的模型解释，就是订阅者订报的过程---订阅者把自己的地址发给报社，报社在确认送报清单里这是个新客户之后，把新客户地址加到地址清单里。
*/
public synchronized void addObserver(Observer o) {
if (o == null)
throw new NullPointerException();
if (!obs.contains(o)) {
obs.addElement(o);
}
}
/**
* 将观察者从被观察者的Vector容器里删除。取消此观察者和被观察者的依赖。
* 订报模型解释—-订阅者通知报社取消订报，报社把该订阅者的地址从地址清单里取消。以后送报就不会送到此订阅者了。
* /
public synchronized void deleteObserver(Observer o) {
obs.removeElement(o);
}
/**
* 下面两个方法是当被观察者状态变化时调用，主要通过for循环依次调用Vector容器存储的Observer（观察者）的update方法，从而通知并更新全部有依赖的观察者。
* 订报模型解释—-新的报纸出版了（状态变化），按照清单地址依次给每个订阅者送报。
*/
public void notifyObservers() {
notifyObservers(null);
}
public void notifyObservers(Object arg) {
Object[] arrLocal;
synchronized (this) {
// if (!changed)
if (!hasChanged())
return;
arrLocal = obs.toArray();
clearChanged();
}
for (int i = arrLocal.length-1; i>=0; i--)
((Observer)arrLocal[i]).update(this, arg);
}
/**
* 被观察者清空所有有依赖的观察者。
* 订报模型解释——报社拖欠工资，送报员把清单烧了。
*/
public synchronized void deleteObservers() {
obs.removeAllElements();
}
/**
* 标记被观察者状态已经改变的事实。在notifyObservers之前调用，可以看notifyObservers代码中changed为false时不会执行for循环通知观察者的方法。
*/
protected synchronized void setChanged() {
changed = true;
}
/**
* 将change状态设置成false，通知观察者方法失效。
*/
protected synchronized void clearChanged() {
changed = false;
}
public synchronized boolean hasChanged() {
return changed;
}
/**
* 返回已添加进Vector容器的观察者的数量。
* 订报模型解释—当前订报人数。
*/
public synchronized int countObservers() {
return obs.size();
}
}
再来看下java.util.Observer，实现观察者的接口。就是订阅者。1
2
3public interface Observer {
void update(Observable o, Object arg);
}
是的，观察者接口就只有这么简单的一个方法，update(Observable o, Object arg)：第一个参数（Observable o）被观察者（主题）本身当做第一个变量，好让观察者知道是哪个被观察者（主题）通知它的；第二个参数（Object arg）传入notifyObservers()的数据对象，如果没有说明则为null。
可以看到，Java里通过Observer接口以及Observable类实现了最基本的观察者模式。通过源码解析我们也可以发现它的实现原理还是非常的简单的，其实是通过维护一个内部数组集合加上接口的回调，却很好的解决了观察者和被观察者解耦的问题。同时，Observable对各方法都加了synchronized关键字(notifyObservers的在方法内)，也就是说，Observable是线程安全的，每次调用方法前都需要取得Observable对象的锁。
3. 总结
在最开始的时候，我们抛出了个概念：“RxJava的异步实现，是通过一种扩展的观察者模式来实现的。”因此这一章节首先带大家回顾了一个基础知识就是常规的观察者模式。我们可以看到，观察者模式是解决对象之间的一对多的依赖问题，而对于RxJava而言，扩展的观察者模式主要解决的是异步实现的问题。这也是我总结出的扩展的观察者模式和常规观察者模式的区别之一。在下一章节会和大家一起学习RxJava的扩展的观察者模式是如何实现的。大家加油。