解密RxSwift核心逻辑

作为ReactiveX家族之一的RxSwift在Github截止现在Star:16K.为什么这个框架如此受欢迎，作为函数响应式框架典型代表，底层实现又是如何实现的呢？这一篇文章全面解密

RxSwift核心流程

RxSwift这个优秀的框架，设计的api也是非常精简，让陌生的用户也能非常快速上手

• 1: 创建序列
• 2: 订阅序列
• 3:发送信号

// 1: 创建序列
_ = Observable<String>.create { (obserber) -> Disposable in
    // 3:发送信号
    obserber.onNext("Cooci -  框架班级")
    return Disposables.create()  // 这个销毁不影响我们这次的解读
    // 2: 订阅序列
    }.subscribe(onNext: { (text) in
        print("订阅到:\(text)")
    })

// 控制台打印：“订阅到:Cooci -  框架班级”

我刚开始在探索的时候，我是比较好奇的：为什么我们的Cooci - 框架班级这个字符串会在订阅序列的subscribe的闭包打印。下面是我的代码分析

分析代码：

• 1：创建序列的代码 Create 后面的 闭包A 里面有 3：发送信号，如果要执行 发送信号 ，必然要来到这个闭包A
• 2：我们执行 2: 订阅序列 跟了 闭包B
• 3：通过结果我们显然知道，先执行 闭包A 把 Cooci - 框架班级 传给了 闭包B
• 猜测：代码里面嵌套了闭包的执行调用！猜测的真实性，我们开始解读源码来验证

PS: 说实话 RxSwift框架的源码的确比较复杂并且很多，很多基础薄弱或者耐性不够的小伙伴很容易放弃。但是你看到这篇博客，你有福了：我会快速简短给你介绍，在最后面会附上我绘制的思维导图！

RxSwift核心逻辑

创建序列

extension ObservableType {
    // MARK: create
    public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
        return AnonymousObservable(subscribe)
    }
}

大家可以很清晰看到我们的 可观察序列 的创建是利用协议拓展功能的create方法实现的，里面创建了一个 AnonymousObservable（匿名可观察序列） 命名还是体现了作者的思维 ：这个类就是一个内部类，具备一些通用特性（具有自己功能的类才会命名） 下面我贴出这个类的继承关系

从上面的图，我们可以清晰的看到的继承关系。那么这么多的内容还有那么多层嵌套，这个地方我们需要掌握什么：

• create 方法的时候创建了一个内部对象 AnonymousObservable
• AnonymousObservable 保存了外界的闭包
• AnonymousObservable继承了 Producer 具有非常重要的方法 subscribe

订阅序列

这里说明这个订阅方法 subscribe 和我们上面所说的 subscribe 不是同一个方法

来自于对 ObservableType 的拓展功能

extension ObservableType {
    public func subscribe(onNext: ((E) -> Void)? = nil, ...) -> Disposable {
           // 因为篇幅 省略不影响我们探索的代码
            let observer = AnonymousObserver<E> { event in                
                switch event {
                case .next(let value):
                    onNext?(value)
                case .error(let error):
                    if let onError = onError {
                        onError(error)
                    }
                    else {
                        Hooks.defaultErrorHandler(callStack, error)
                    }
                    disposable.dispose()
                case .completed:
                    onCompleted?()
                    disposable.dispose()
                }
            }
            return Disposables.create(
                self.asObservable().subscribe(observer),
                disposable
            )
    }
}

代码说明：

• E 这里的意思是 Swift 的关联类型，这个如果仔细看过可观察序列的继承链源码应该不难得出：这个E 就是我们的 序列类型，我们这里就是String

public class Observable<Element> : ObservableType {
    /// Type of elements in sequence.
    public typealias E = Element

• 创建了一个 AnonymousObserver (匿名内部观察者) 手法和我们的 AnonymousObservable 差不多，它这里的初始化是闭包参数，保存了外界的 onNext, onError , onCompleted , onDisposed 的处理回调闭包的调用,下面我还是给大家贴出 观察者 的继承链关系，帮助大家理解

• self.asObservable() 这个是我们的 RxSwift 为了保持一致性的写法
• self.asObservable().subscribe(observer)其实本质就是 self.subscribe(observer),通过可观察序列的继承关系，我们可以非常快速的定位 Producer 订阅代码

override func subscribe(_ observer: O) -> Disposable where O.E == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // 篇幅原因，我们省略一些代码，方便我们理解
            ...
            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
              // 篇幅原因，我们省略一些代码，方便我们理解
              ...
                return disposer
            }
        }
    }

• 关于销毁代码和调度者代码这里不分析
• self.run 这个代码最终由我们生产者 Producer 延伸到我们具体的事务代码 AnonymousObservable.run

override func run (...) {
    let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
    let subscription = sink.run(self)
    return (sink: sink, subscription: subscription)
}

• sink.run的写法也是比较好的，业务处理的还是下沉了，让分工更加明确

func run(_ parent: Parent) -> Disposable {
    return parent._subscribeHandler(AnyObserver(self))
}

• parent 就是上面传过来的AnonymousObservable对象
• 我们非常兴奋的看到 AnonymousObservable._subscribeHandler,从这句代码我们解惑了为什么我们的序列订阅的时候流程会执行我们 序列闭包，然后去执行 发送响应
• 发送响应的代码等会分析，这里还有一个比较重要的家伙 AnyObserver(self)

public init<O : ObserverType>(_ observer: O) where O.E == Element {
    self.observer = observer.on
}

• 在这个构造方法里面，我们创建了一个结构体 AnyObserver 保存了一个信息 AnonymousObservableSink .on 函数，不是 AnonymousObservableSink,这个地方一般初次来到这里的人都会犯错误。不知道你是否意识到了！

发送响应

我们从上面的分析，非常清晰:
obserber.onNext("Cooci - 框架班级") 的本质是： AnyObserver.onNext("Cooci - 框架班级")

这时候发现我们的AnyObserver 是没有这个方法，这很正常！一般思路，找父类，找协议

extension ObserverType {
    public func onNext(_ element: E) {
        self.on(.next(element))
    }
}

• 外界 obserber.onNext("Cooci - 框架班级") 再次变形 ：AnyObserver.on(.next("Cooci - 框架班级")) ,这里大家一定要主要，这个AnyObserver调用了 on 里面传的是 .next函数, .next函数带有我们最终的参数

public struct AnyObserver<Element> : ObserverType {
    public init<O : ObserverType>(_ observer: O) where O.E == Element {
        self.observer = observer.on
    }
    public func on(_ event: Event<Element>) {
        return self.observer(event)
    }
}

• self.observer 构造初始化就是：AnonymousObservableSink .on 函数
• 看到这里又要变形咯：self.observer(event) -> AnonymousObservableSink .on(event) 其中 event = .next("Cooci - 框架班级") 最终我们的核心逻辑又回到了 sink 这个神奇的管子，看到这里不禁拍案叫绝， RxSwift这个设计能力，还有谁~~~

class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
    func on(_ event: Event<E>) {
        switch event {
        case .next:
            if load(self._isStopped) == 1 {
                return
            }
            self.forwardOn(event)
        case .error, .completed:
            if fetchOr(self._isStopped, 1) == 0 {
                self.forwardOn(event)
                self.dispose()
            }
        }
    }
}

• self.forwardOn(event) 这也是执行的核心代码，因为 AnonymousObservableSink 继承 Sink 这里还有封装，请看下面的代码

class Sink<O : ObserverType> : Disposable {
    final func forwardOn(_ event: Event<O.E>) {
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
}

• 其中 self._observer 就是我们初始化保存的 观察者：AnonymousObserver
• 那么我们变形得出本质就是：AnonymousObserver.on（.next("Cooci - 框架班级")）,我的天啊! 这里逻辑辗转回到了我们 订阅序列 时候创建的 AnonymousObserver 的参数闭包的调用！所有的一切感觉是这样的啰嗦，但又是这么的顺其资源。

let observer = AnonymousObserver<E> { event in
    switch event {
    case .next(let value):
        onNext?(value)
    case .error(let error):
        if let onError = onError {
            onError(error)
        }
        else {
            Hooks.defaultErrorHandler(callStack, error)
        }
        disposable.dispose()
    case .completed:
        onCompleted?()
        disposable.dispose()
    }
}

• 判断 event 进而调用 onNext?(value) ,因为枚举的关联值（Swift很强大的功能）value = "Cooci - 框架班级", 接下来就是外界 onNext闭包的调用传参，那么这个时候源码解析到这里，我相信你已经完全掌握了RxSwift的核心逻辑，最后这里附上我们的分析图解

总结：RxSwift的结构

• 1：就是序列感念 满世界都是序列 - 编码统一 ，随时随地享用
• 2：通过函数式思想吧一些列的需求操作下沉（把开发者不关心的东西封装） - 优化代码，节省逻辑
• 3：RxSwift最典型的特色就是解决Swift这门静态语言的响应能力，利用随时间维度序列变化为轴线，用户订阅关心能随轴线一直保活，达到订阅一次，响应一直持续~