RxSwift 核心原理解析
RxSwift 核心原理解析
角色定位
- 观察者（Observer）
- 被观察者（Observable）
- 订阅者（Subscriber） 事件的最终处理者
- 管道（Sink） Observer 和 Observable 沟通的桥梁
Simple Example
Observable<String>.create { observer -> Disposable in
observer.onNext("hello")
return Disposables.create()
}
.subscribe { event in
print(event.element)
}
我们重点关注：
create 闭包什么时候执行，
subscribe 闭包又是什么时候执行的
Class Diagram
Observable.png
Observable的核心函数:
-
subscribe订阅这个大家都知道 -
run对用户不可见，隐藏了大量的实现细节 -
asObservable： 看似比较鸡肋，实际上作用很大。举个简单例子：
某个具体Observable类的重载方法
/// Converts `self` to `Observable` sequence.
public func asObservable() -> Observable<Element> {
return source
}
ObservableType 协议扩展
extension ObservableType {
/**
Subscribes an event handler to an observable sequence.
- parameter on: Action to invoke for each event in the observable sequence.
- returns: Subscription object used to unsubscribe from the observable sequence.
*/
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
}
-
asObservable: 可以让我们使用组合方式去组装类，不需要自己每次亲力亲为的实现subscribe - 有利于保持代码的简洁性，
return self.asObservable().subscribe(observer)就是良好的封装性的体现
ObserverType.png
Observer 核心函数：
-
on: 事件发生器 -
onCore: 对用户不可见，隐藏具体实现细节
Sink.png
Sink 核心函数：
- forwardOn
- on
- run
代码分析
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
先看看create 函数， 返回一个AnonymousObservable实例，这是因为Observable是无法正常工作的，它内部有不少抽象方法没有实现
// 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)
}
}
首先说一下继承体系： AnonymousObservable -> Producer -> Observable -> ObservableType -> ObservableConvertibleType
每一层都只处理一点点事情，剩下的交给下一层处理
ObservableConvertibleType： 完全的抽象
ObservableType： 处理subscribe
Observable: 处理 asObservable
Producer: 重载subscribe
AnonymousObservable： 处理run
也就是说如果说如果我们要自定义一个Observable的话，通常只需要继承Producer， 并实现run方法。AnonymousObservable做的事情也不多，实现run方法，作为create闭包的持有者。
run方法涉及另外一个类AnonymousObservableSink,Sink作为Observer 和 Observable的衔接的桥梁，之前还在想为什么叫做ObservableSink，现在想明白了。Sink本身遵守ObseverType协议，与此同时实现了run方法，虽然没有实现subscribe方法，但是已经足够了，这样sink从某种程度来说也是Observable，通过sink就可以完成从Observable到Obsever的转变。
subscribe:
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
这里不太明白为什么一定要写成这样：
let observer = AnonymousObserver { e in
on(e)
}
感觉直接写成let observer = AnonymousObserver(on) 会更加清晰。AnonymousObserver 比较简单持有subscribe闭包，并且实现onCore method.
刚才的只是个入口，这个才是核心
// 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
}
}
}
这个if 和 else 不用管这个是线程安全相关的，不是我关注的重点。这里返回的是 Disposable , Disposable 跟对象的生命周期密切相关。暂且在这打个tag吧。
- branch_1: 线程安全
- branch_2: 生命周期管理
- branch_3: 核心逻辑实现细节
好了我们现在关注的是核心逻辑实现细节，其他有时间再补上。核心语句
let sinkAndSubscription = self.run(observer, cancel: disposer)
我们来看看run的逻辑
// 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("hello")
return Disposables.create()
}
也就是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
}
}
构造函数的入参是EventHandle,结合上下文 AnyObserver.observer = AnonymousObservableSink.on,那么这下就明白了 observer.onNext("hello") 最终会触发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()
}
}
}
老规矩忽略线程安全问题，关注核心问题
// 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)
}
我们最初传进来的subscribe闭包的实体类，理解了这个接下来就简单了
// 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)
}
}
}
// AnonymousObserver.onCore
override func onCore(_ event: Event<Element>) {
return _eventHandler(event)
}
这里的_eventHandler 就是最初传进来的订阅闭包即：
.subscribe { event in
print(event.element)
}
好了到此为止了，总结一下：
来个堆栈看的清晰些：
Rx subscribe源码分析.png
再来一个类图
Compose Diagram.png
稍微解释一下整个过程分为两个阶段：
- Obsevable 构建阶段，这里使用create构造方法构造Obsevable，还有其他各种各样的构造方法，这里不一一赘述。
- subscribe 阶段， Obsevable.subscribe ---> Obsevable.run ----> Sink.run ----> AnyObserver.On ----> Sink.on ----> Observer.On ---> Observer.OnCore ----> eventHandler
这个过程也可以分为多个阶段：
Obsevable.subscribe ---> Obsevable.run ----> Sink.run 这个过程我理解为build阶段，通过Sink建立Obsevable和 Observer的联系，在这个阶段Sink扮演还是Obsevable角色。
AnyObserver.On ----> Sink.on ----> Observer.On ---> Observer.OnCore 这个阶段的话Sink扮演的就是 Observer 角色
