一.标准的观察者模式

生活中的例子

标准的观察者模式，生活当中用的很多，比如微信公众号

有三个人，和一个微信公众号。人可以订阅微信公众号，微信公众号发生改变（比如推送新文章）的时候会通知人。也就是微信公众号是被观察者，人是观察者。
还有一个突出的特点是：一个被观察者，多个观察者。

对应到程序

对应到程序上，可以用这幅图来表示

写程序的时候就可以按图那样先写出图中显示的四种类，然后在被观察者的接口（或者其实现类）中，定义一个容器，可以是集合或者数组，里面装着所有对观察者的引用。当被观察者发生改变的时候，就依次通知观察者，这样就实现了观察者设计模式。

二.RxJava的Hook机制

1.Hook简介

Hook的意思是“钩、钩子”。Hook机制的意思就是说，当任务在往下执行的时候，我先把你勾出来，先执行下我的代码，然后再把你送回去继续执行你自己的。所以Hook有时也是可以搞破坏的。鉴于Hook的这种功能特点，一般都是用到全局的静态的变量或方法。

2.RxJava的Hook机制。

先看一段最普通的RxJava的代码

Observable.create(new ObservableOnSubscribe<Object>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Exception {emitter.onNext("A");}}).map(new Function<Object, Boolean>() {@Overridepublic Boolean apply(@NonNull Object o) throws Exception {return false;}}).subscribe(new Observer<Boolean>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {}@Overridepublic void onNext(@NonNull Boolean aBoolean) {}@Overridepublic void onError(@NonNull Throwable e) {}@Overridepublic void onComplete() {}});

在订阅，也就是subscribe之前，执行了两个方法，分别是create和map，让我们都进去看一下这两个方法。

create

map

可见这两个方法都执行了红框的部分，我们点进去onAssembly看一下，

onAssembly

看到这，可以发现onAssembly方法相当于啥也没做，传进来source，然后又return了回去。看似没用，实则可以理解成是为我们准备的。我们可以调用这个方法，这样在执行create或map或其他方法的时候都会调用一下我们调用的全局方法，也就是让f不为null，也就是设置onObservableAssembly的值，我们看到只有这一个方法满足要求

所以我们调用这个方法为onObservableAssembly赋值，在onCreate方法中

//Hook
//以下这段代码与正常的相比就是多了Log日志打印。
RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {@Overridepublic Observable apply(@NonNull Observable observable) throws Exception {Log.v("ljh","整个项目全局监听，调用了多少次这个方法："+observable);return observable;//注意这里是返回observable而不是null}
});

程序执行的结果是：

3.为什么打印出这两个对象

为什么是这两个对象呢，其实就是下面的create方法和map方法返回的对象。我们就看create方法吧，

①进入create方法

它new了一个ObservableCreate对象，作为参数传入了onAssembly方法，我们

②进入onAssembly方法

③进入apply方法

可以看到它调用了f的apply方法，并把ObservableCreate对象作为参数传进去。

④我们追踪进入apply方法

可以看到它是一个接口，实现其实就在Hook的时候setOnObservableAssembly方法传入的参数里面，

这个observable就是ObservableCreate对象。ObservableMap也一样，这里就不再赘述。重点是理解Hook的思想。

4.怎么利用Hook做破坏？（仅限理解，不要这么做！！！）

如果想做破坏的话，也很简单，就是把返回observable改成返回null，这样本来是ObservableCreate.map，变成了null.map，必然会出现空指针异常的错误。如图

而且从源码也可以看出来onObservableAssembly是全局静态的，也符合Hook的特征

二.RxJava的观察者模式

1.正常的RxJava代码的执行流程源码分析

我们先分析下一个正常的RxJava代码的执行流程
以这个标准的RxJava为例进行说明

Observable.create(new ObservableOnSubscribe<Object>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Object> emitter) throws Exception {emitter.onNext("A");}}).subscribe(new Observer<Object>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {}@Overridepublic void onNext(@NonNull Object object) {}@Overridepublic void onError(@NonNull Throwable e) {}@Overridepublic void onComplete() {}});

1.Observer

也就是这一部分

我们给其命名为自定义观察者，ok，进入源码

它是一个接口，值得一提的是这个泛型，它的实现就是我们自己实现的。在上面的标准RxJava示例中泛型为Object，那么这里也是Object，也就是onNext里面的参数也是Object类型，很简单。

2.Observable的创建过程

也就是这一部分

我们给红框里面的部分命名为自定义source。在create方法里面，我们把**自定义source**传进去了

我们进入ObservableCreate看一下

发现它在构造方法里面把我们的 自定义source 赋值给了它的source变量

创建过程的时序图如下

3.subscribe订阅

也就是这一部分

这一部分是最复杂的，当然，掌握了这一部分，也就知道为什么说RxJava是U型结构了。

①我们首先进入subscribe

它把我们的自定义观察者作为参数传进去了，如上图和下图

②可以发现它调用了subscribeActual，我们进入

发现它是一个抽象方法，那么到底是在哪调用的呢？我们想到了在执行create方法之后，返回的是ObservableCreate对象，然后它调用的subscribe方法，（ObservableMap里面没有显示的subscribe方法）。所以

③我们进去ObservableCreate看

发现果然有！这个方法首先是定义了一个发射器，并把自定义观察者作为参数传入，然后调用了自定义观察者的onSubscribe方法（这也解释了为什么一订阅就执行onSubscribe方法），然后执行了自定义source的subscribe方法，让我们

④回看一下自定义source

白色部分是自定义source，红框部分是自定义source的subscribe方法，也就是调到了这里，同时也拿到了发射器。然后执行发射器的onNext方法。而发射器正好又有自定义观察者，所以又调回了自定义观察者的onNext（当然，如果有map就调用map）。

这样的调用流程就是先从自定义观察者调用到自定义source，然后又调回自定义观察者，画出图就是一个U状，所以说RxJava是一个U型结构

订阅流程时序图

2.RxJava的观察者模式和一般的观察者模式区别

直观分析

• 标准的观察者模式是一个被观察者，多个观察者，并且需要被观察者主动发出改变通知，所有的观察者才能观察到。
• RxJava的观察者模式是多个被观察者，一个观察者，并且需要起点和终点订阅后，终点才能观察到被观察者。

为什么说是多个被观察者呢？看create和map


返回的都是Observable，也就是被观察者，只有终点一个观察者，也就是我们前面常说的自定义观察者。

从模式角度分析

• 在标准的观察者设计模式中，当发出通知改变时，会遍历Observable里面的容器，此容器里面有10个Observer，就会通知10个Observer
• 在RxJava观察者设计模式中，分发事件时，会拿到发射器，通过发射器关联到我们自定义的Observer，发射器调用到我们自定义的Observer
  画图来看的话就是这样
  

当增加一个map方法的时候，直接将发射器发射给map就好了，极大地降低了耦合度。