RxSwift 核心逻辑-源码分析

示例代码

要想了解RxSwift核心逻辑，首先要知道他的运行流程，运行示例代码，看一下输出结果

        let ob = Observable<Any>.create { (observer) -> Disposable in
            observer.onNext("发送信号")
            observer.onCompleted()
            return Disposables.create()
        }
        let _ = ob.subscribe(onNext: { (text) in
            print("订阅到:\(text)")
        }, onError: { (error) in
            print("error: \(error)")
        }, onCompleted: {
            print("完成")
        }) {
            print("销毁")
        }

输出日志

订阅到:发送信号
完成
销毁

创建序列

首先点击create跟一下流程

image.png
发现当前类中，并没有找到具体实现,一顿操作，在Create中找到实现，返回AnonymousObservable(subscribe),注意subscribe参数：
image.png
在AnonymousObservable中可以看到，将传进来的闭包定义别名SubscribeHandler，同时通过属性_subscribeHandler将闭包保存起来:
保存闭包

订阅

• 紧接着订阅信号点击发现有两个方法，因为Observable遵循协议ObservableType,所以进入第二个查看

public class Observable<Element> : ObservableType {...}

public protocol ObservableType : ObservableConvertibleType {...}

订阅信号0
subscribe方法实现源码如下：首先实例化observer
保存闭包

在这里传入参数onNext,onError,onComplete, onDisposed四个闭包，通过AnonymousObserver同时订阅四个闭包，当外界调用者传递一个参数，就会响应AnonymousObserver中对应的闭包event，实现监听。

订阅信号1

• Disposables.create，创建垃圾袋，这是垃圾回收机制，所以暂时不关心，重点在self.asObservable().subscribe(observer), disposable )，如何理解？此时的self还记得是谁吗，create的时候返回是AnonymousObservable,

 return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )

在AnonymousObservable中并没有找到asObservable(),通过继承关系，AnonymousObservable ->Producer-> Observable,在父类Observable中找到实现,本质上就是返回了自身，代码如下,

image.png
继续查找AnonymousObservable .subscribe(observer)同上查找方法，在父类Producer中找到了subscribe方法，如下：
AnonymousObservable .subscribe(observer)
分析可知，两种情况都会调用代码run，在Producer中定义了run代码如下：
Producer.run()
但是发现什么都没做，这个时候，还是回归到刚刚的self既AnonymousObservable中查找，
AnonymousObservable.run

此时实例化管线sink，在父类中保存observer

Sink.init

然后调用run,查看AnonymousObservableSink代码如下：

AnonymousObservableSink

调用run，其中parent是上图中定义的别名，实际上是AnonymousObservable，还记得开始创建序列的时候，AnonymousObservable中保存的闭包_subscribeHandler吗，没错就是他！接下来分析AnyObserver(self)既AnyObserver(AnonymousObservableSink)，也就是结构体初始化传递了参数,self.observer = observer.on，分析得知AnonymousObservableSink.on函数赋值给observer.on；
从这里就可以明白为什么这行代码observer.onNext(“发送信号”) 最终会触发AnonymousObservableSink.on事件。

image.png

发送信号

开始调用observer.onNext("发送信号"),首先进入

发送信号

接下来，瞄一眼AnonymousObservableSink.on源码，在接收到事件时，会调用self.forwardOn(event),

AnonymousObservableSink.on

跟踪self.forwardOn(event)如下：这里的self._observer就是之前调用subscribe函数里面创建的observer对象。

self.forwardOn(event)

调用父类ObserverBase.on,然后调用onCore，

image.png

上图中，onCore并没有具体实现，所以在子类中查找如下图

响应闭包
还记得这张图吗？没错就是开始订阅信号的时候，保存的闭包，现在开始响应了，至此就实现了响应，最后在Disposables中销毁。