RXSwift-核心逻辑
基本概念
要想充分理解RXSwift核心逻辑，那么首先必须要知道RXSwift里包含哪几个角色，以及它们的职责。
被观察者(Observable)
它主要负责产生事件，实质上就是一个可被监听的序列(Sequence)。
Observable<T>这个类就是Rx框架的基础，我们称它为可观察序列。
Observable<T>
==异步产生==>event(element : T)
被观察者(Observable)类继承关系
观察者(Observer)
它主要负责监听事件然后对这个事件做出响应，或者说任何响应事件的行为都是观察者。
观察者(Observer)类继承关系
订阅者(Subscriber)
事件的最终处理者
管道(Sink)
Observer 和 Observable 沟通的桥梁：Sink相当与一个加工者，可以将源事件流转换成一个新的事件流，如果将事件流比作水流，事件的传递过程比作水管，那么Sink就相当于水管中的一个转换头。
管道(Sink)类继承关系
代码解析
接下来我们结合以下很简单的代码来分析，逐步揭开RXSwift的神秘面纱。
//1:创建序列
let ob = Observable<Any>.create { (observer) -> Disposable in
// 3:发送信号
obserber.onNext("测试OnNext")
obserber.onCompleted()
obserber.onError(NSError.init(domain: "error！", code: 000, userInfo: nil))
return Disposables.create()
//2:订阅信息
let _ = ob.subscribe(onNext: { (text) in
print("订阅到:\(text)") //text从哪里来的？
}, onError: { (error) in
print("error: \(error)") //error从哪里来的？
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
在这里我们主要关注下
create 闭包什么时候执行，
subscribe 闭包又是什么时候执行的
也就是3->2这条线
Create方法
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
我们看到create 函数， 返回了一个AnonymousObservable实例，接着我们进入AnonymousObservable看它里面具体内容
// AnonymousObservable Class
final fileprivate class AnonymousObservable<Element> : Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
_subscribeHandler = subscribeHandler
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
在这里我们再次回顾下Observable集成体系
（父类）
ObservableConvertibleType（完全的抽象）
|
ObservableType（ 处理subscribe）
|
Observable（处理 asObservable）
|
Producer（重载subscribe）
|
AnonymousObservable（处理run）
（子类）
PS：由上面我们能看出，如果想自定义Observable通常只需要继承Producer， 并实现run方法就可以了。
在这里我们看到Run方法里面涉及了类AnonymousObservableSink,它作为Observer 和 Observable的衔接的桥梁我们看到它本身遵守ObseverType协议，与此同时实现了run方法。
那也就是说，
sink从某种程度来说也是Observable
通过sink就可以完成从Observable到Obsever的转变。
总结下create方法主要工作：
- 创建
AnonymousObservable对象， - 用
_subscribeHandler保存了闭包 - 写了
run方法在内部创建了AnonymousObservableSink
Subscribe方法
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
上述代码只是入口，下面的才是核心（Producer里面）
// Producer Class
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
if !CurrentThreadScheduler.isScheduleRequired {
// The returned disposable needs to release all references once it was disposed.
let disposer = SinkDisposer()
let sinkAndSubscription = run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
else {
return CurrentThreadScheduler.instance.schedule(()) { _ in
let disposer = SinkDisposer()
let sinkAndSubscription = self.run(observer, cancel: disposer)
disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)
return disposer
}
}
}
这里面我们看到了Producer调用了自己的run方法，而AnonymousObservableSink作为其子类重写了该方法，我们先去看下子类是如何写的。
// AnonymousObservable Class
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
之前提到过AnonymousObservableSink，注意Sink是持有Observer的，从这也可以看出来
Observerable的run方法触发Sink的run方法
接下来就要关注AnonymousObservableSink方法。
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
在这里我们再一次见到了subscribeHandler，这个subscribeHandler就是之前最开始的闭包！
Observable<String>.create { observer -> Disposable in
observer.onNext("测试")
return Disposables.create()
}
至此我们知道了create闭包是什么执行的了，接下来我们自然的把目光锁定到实体类AnyObserver，看看它里面究竟是如何实现的。
public struct AnyObserver<Element> : ObserverType {
/// The type of elements in sequence that observer can observe.
public typealias E = Element
/// Anonymous event handler type.
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
/// Construct an instance whose `on(event)` calls `eventHandler(event)`
///
/// - parameter eventHandler: Event handler that observes sequences events.
public init(eventHandler: @escaping EventHandler) {
self.observer = eventHandler
}
/// Construct an instance whose `on(event)` calls `observer.on(event)`
///
/// - parameter observer: Observer that receives sequence events.
public init<O : ObserverType>(_ observer: O) where O.E == Element {
self.observer = observer.on
}
/// Send `event` to this observer.
///
/// - parameter event: Event instance.
public func on(_ event: Event<Element>) {
return self.observer(event)
}
/// Erases type of observer and returns canonical observer.
///
/// - returns: type erased observer.
public func asObserver() -> AnyObserver<E> {
return self
}
}
在这里我们看到其内部的Observer其实是一个EventHandler，并且在初始化的时候把外部传过来的AnonymousObservableSink.on赋值给了这个Observer，也就是说observer.onNext("测试")最终会触发AnonymousObservableSink.on事件
接着我们看下AnonymousObservableSink的on的具体实现
// AnonymousObservableSink.on
func on(_ event: Event<E>) {
#if DEBUG
_synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { _synchronizationTracker.unregister() }
#endif
switch event {
case .next:
if _isStopped == 1 {
return
}
forwardOn(event)
case .error, .completed:
if AtomicCompareAndSwap(0, 1, &_isStopped) {
forwardOn(event)
dispose()
}
}
}
AnonymousObservableSink的on会调用其内部的forwardOn
// Sink.forwardOn
final func forwardOn(_ event: Event<O.E>) {
#if DEBUG
_synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { _synchronizationTracker.unregister() }
#endif
if _disposed {
return
}
_observer.on(event)
}
在这里还记得这个_observer的实体吗，看到下面代码你一定会回忆起来，
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
return self.asObservable().subscribe(observer)
}
对，没错，这个_observer就是我们最初传进来的subscribe闭包的实体类AnonymousObserver~!
接着我们继续看下这个AnonymousObserver的on方法又是如何实现的
继承关系：
AnonymousObserver->ObserverBase->ObserverType
// ObserverBase.on
func on(_ event: Event<E>) {
switch event {
case .next:
if _isStopped == 0 {
onCore(event)
}
case .error, .completed:
if AtomicCompareAndSwap(0, 1, &_isStopped) {
onCore(event)
}
}
}
ObserverBase.on会触发onCore方法，看下子类的实现
// AnonymousObserver.onCore
override func onCore(_ event: Event<Element>) {
return _eventHandler(event)
}
这里的_eventHandler还记得吗？它就是最初传进来的订阅闭包即：
.subscribe { event in
print(event.element)
}
总结一下
Obsevable-Create阶段：
- 创建
AnonymousObservable - 保存闭包(
subscribeHandler)
Obsevable-Subscribe阶段：
- 创建
AnonymousObserver - 调用自己（
AnonymousObservable）的run方法:（AnonymousObserver作为参数）
AnonymousObservable重写了run，它在方法里面创建了AnonymousObservableSink并在sink里保存了这个刚创建的AnonymousObserver - 调用
AnonymousObservableSink的run：run方法里用到AnonymousObservable的_subscribeHandler并传入AnyObserver，这里AnonymousObservableSink.on赋值给了AnyObserver内部的EventHandler成员observer
执行阶段：
AnyObserver.on ----> AnonymousObservableSink.on ----> AnonymousObservableSink.forwardOn ----> AnonymousObserver.on---> AnonymousObserver.onCore ----> _eventHandler(event)
PS：
可以看出Sink在不同的阶段有着不同的身份
- Sink充当Obsevable
Obsevable.subscribe--->Obsevable.run---->Sink.run
这个过程通过Sink建立Obsevable和 Observer的联系
- Sink充当Observer
AnyObserver.on---->Sink.on---->Observer.on--->Observer.OnCore
