大家好，欢迎来到IT知识分享网。

//最初创建的source
val source = object : ObservableOnSubscribe<String> {
    override fun subscribe(emitter: ObservableEmitter<String>) {
        emitter.onNext("文章1")
        emitter.onNext("文章2")
    }
}
//会返回包含source的ObservableCreate对象
val observable1 = Observable.create<String>(source)

//会返回包含observable1的ObservableMap
val observable2 = observable1.map(object : Function<String, String> {
    override fun apply(t: String): String {
        return t + "111111111111"
    }
})

//会返回包含observable2的ObservableFilter
val observable3 = observable2.filter(object : io.reactivex.functions.Predicate<String>{
    override fun test(t: String): Boolean {
        return t.contains("2")
    }
})

//步骤2. 创建观察者(Observer)，接受事件并做出响应操作。
val observer: Observer<String> = object : Observer<String> {
    override fun onSubscribe(d: Disposable) {
        showLog("onSubscribe")
    }

    override fun onNext(s: String) {
        showLog("onNext : $s")
    }
}

//步骤3. 观察者通过订阅（subscribe）被观察者把它们连接到一起。
observable3.subscribe(observer)

第一阶段：

当执行到步骤3时的对象结构图为：

图（一）

从这个结构图可以看出，对Observable添加的各种操作符，其实就是一个封包裹的过程。

最里面的source为@20459，observer并没有变化。

第二阶段：

再来看看执行 observable3.subscribe(observer) 发生了什么

此时的最外层是ObservableFilter，所以会执行ObservableFilter的subscribeActual

 注意，这里的source就是上面的observable2，即：ObservableMap。接着调用

source.subscribe(new FilterObserver<T>(observer, predicate));

就会执行ObservableMap的subscribeActual

 这里的source就是observable1，即：ObservableCreate，接着就是执行ObservableCreate的subscribeActual

图（二）

这里的source就是@20459

执行到这里时，我们的调试堆栈就变成了这样：

 而上面图（一）的数据结构图就变成了下面这样：

图（三）

从图中可以看出，之前（ObservableCreate）封的包裹被拆开了，而observer却被包裹起来了。

第三阶段：

接着 图（二）往下执行，就是observer拆包裹的过程，因为source就是我们最初创建的source，如下图：

 所以 source.subscribe(parent) ，执行的就是上图中的subscribe，parent传给了emitter。这里的parent就是图（三）中的parent（CreateEmitter）

emitter.onNext("文章1") //执行CreateEmitter的onNext()

 由图（三）可以知道这里的observer就是MapObserver，所以就会去执行MapObserver的onNext()

这里首先会执行mapper变换，就是我们之前添加的

 super(actual)会执行

 所以这里的downstream就是传进来的t，就是图（三）中的FilterObserver，所以：

downstream.onNext(t)就会执行FilterObserver类中的onNext()函数

首先会执行filter.test()，也就是我们添加的

 这里的downstream由图（三）可知是我们包裹最里面的observer，也就是最终接收的observer，downstream.onNext(t)就会执行

 至此，流程就走完了。最后画个流程图总结一下：