原文地址：https://www.jianshu.com/p/0e0703466483
作为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核心逻辑
创建序列
代码语言：txt复制extension ObservableType {
// MARK: create
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
}大家可以很清晰看到我们的 可观察序列 的创建是利用协议拓展功能的create方法实现的，里面创建了一个 AnonymousObservable（匿名可观察序列） 命名还是体现了作者的思维 ：这个类就是一个内部类，具备一些通用特性（具有自己功能的类才会命名） 下面我贴出这个类的继承关系
从上面的图，我们可以清晰的看到的继承关系。那么这么多的内容还有那么多层嵌套，这个地方我们需要掌握什么：
create方法的时候创建了一个内部对象AnonymousObservableAnonymousObservable保存了外界的闭包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这门静态语言的响应能力，利用随时间维度序列变化为轴线，用户订阅关心能随轴线一直保活，达到订阅一次，响应一直持续~
