今天带大家解读下订阅发布流程的内部源码。
本系列文章RxSwift使用的版本是：6.2.0
let observable = Observable<Int>.create { (anyObserver) -> Disposable in
anyObserver.onNext(1)
anyObserver.onCompleted()
return Disposables.create ()
}
let dispose = observable.subscribe(onNext: { ele in
print(ele)
}, onDisposed: {
print("dispose")
})
dispose.dispose()
上面代码创建一个被观察者，并订阅它，打印序列元素，最后dispose，最后打印：
1
dispose
RxSwift的基本订阅流程涉及到许多类和协议，需要先弄清楚各个类和协议的作用以及它们之间的关系，否则直接看代码容易绕晕。
先上个类图把所有相关类和协议关系理清楚。
类图.jpg
其中蓝色的是类，橙色的协议，黄色的是枚举， 绿色的是结构体。我们一个一个说：
- 协议：
ObservableConvertibleType这是个被观察者的协议，唯一的一个协议方法是asObservable(), 表示可以转换成被观察者。
ObservableType： 继承自ObservableConvertibleType，可以创建被观察者，任何被观察者类需要遵循此协议。在extension中定义了一个create函数用来创建被观察者，两个subscribe函数用来订阅观察者。
Disposable订阅取消的接口，只有一个disposes协议方法，用来释放相关资源。
Cancelable继承自Disposable，有一个isDisposed协议方法，表示是否已释放资源。
ObserverType表示序列发布者，可以发送序列。其中有几个我们比较熟悉的方法：onNext onError onCompleted 这三个会调用on方法，on只是个协议方法，看下代码就明白了：
/// Supports push-style iteration over an observable sequence.
public protocol ObserverType {
/// The type of elements in sequence that observer can observe.
associatedtype Element
@available(*, deprecated, message: "Use `Element` instead.")
typealias E = Element
/// Notify observer about sequence event.
///
/// - parameter event: Event that occurred.
func on(_ event: Event<Element>)
}
/// Convenience API extensions to provide alternate next, error, completed events
extension ObserverType {
/// Convenience method equivalent to `on(.next(element: Element))`
///
/// - parameter element: Next element to send to observer(s)
public func onNext(_ element: Element) {
self.on(.next(element))
}
/// Convenience method equivalent to `on(.completed)`
public func onCompleted() {
self.on(.completed)
}
/// Convenience method equivalent to `on(.error(Swift.Error))`
/// - parameter error: Swift.Error to send to observer(s)
public func onError(_ error: Swift.Error) {
self.on(.error(error))
}
}
onNext onError onCompleted都调用了on方法，它还包含一个关联类型，Element可以认为是个范型，表示元素的类型。
- 类
ObservableProducerAnonymousObservable这三个都是被观察者，依次继承的关系，Observable遵循ObservableType， Producer 继承Observable，AnonymousObservable继承了Producer。Producer，实现subscribe，run方法， AnonymousObservable实现了run方法。
Sink 和 AnonymousObservableSink 这两个算是是整个流程的核心类，对消息订阅发送进行管理。其中Sink 遵循Dispsable，包含两个属性ObserverType和Cancelable，这是整个类图唯一的两个组合关系，其他类的属性都是闭包，AnonymousObservableSink继承了Sink，同时遵循了ObserverType，也就是说AnonymousObservableSink 包括一个ObserverType类型的属性（由父类继承而来）同时又遵循了ObserverType协议，这让我想起的设计模式中的装饰模式，其实整个Rx框架还有很多类型的Sink。
ObserverBase AnonymousObserver, 遵循了Dispose和 ObserverType，这两个是观察者，继承关系。
-
结构体 唯一的一个结构体AnyObserver，遵循了ObserverType 这个好像也是观察者，实际上是序列的发送者，用户通过它来调用onNext 等方法发送序列，被观察者如Observable通过AnyObserver发送序列，而AnonymousObserver对象负责接收序列。
-
枚举 Event 表示序列事件，包含next(Element) error(Swift.Error) onCompleted三个case。
终于介绍完了所有的类和协议。
接下来需要说到三个重要的闭包：
- 创建被观察者的闭包即我们一开始的代码中的
{ (anyObserver) -> Disposable in
anyObserver.onNext(1)
anyObserver.onCompleted()
return Disposables.create ()
}
它的类型是（AnyObserver）-> Disposeable 在订阅时会执行这个闭包，并且用anyObserver发送序列。
- 订阅时的onNext的闭包：
(onNext: { ele in
print(ele)
}, onDisposed: {
print("dispose")
})
这样发送next序列时，会调用这个闭包。
- 第三个闭包在subscribe方法的内部，我们进去看一下：
let observer = AnonymousObserver<Element> { event in
#if DEBUG
synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { synchronizationTracker.unregister() }
#endif
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()
}
}
这里创建了一个AnonymousObserver， 也就是观察者，创建时将闭包作为初始化参数，闭包中调用了第二个闭包onNext。所以第一个闭包和第三个闭包是如果关联的呢？也就是订阅的时候如何调用第一个闭包，第一个闭包再调用到第三个闭包？这是整个流程的关键？
我们再上个流程图看看整个订阅发布的流程。
订阅流程图.jpg
根据这个图再结合源码一步一步分析下流程。
- 首先通过Observable的 create方法创建序列，ObservableType extension 提供了实现，而Observable遵循了ObservableType协议，可以看到代码创建了AnonymousObservable(subscribe), 并把闭包传进入，而AnonymousObservable类内部持有了这个闭包。
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
return AnonymousObservable(subscribe)
}
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
}
- 用返回的AnonymousObservable对象，调用subscribe方法，这个方法在ObservableType 扩展中，看一看subscribe主要代码：
let observer = AnonymousObserver<Element> { event in
#if DEBUG
synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { synchronizationTracker.unregister() }
#endif
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
)
先包装一个AnonymousObserver，即观察者，把闭包作为初始化方法的参数传进去，并赋值给它的_eventHandler属性，跟AnonymousObservable比较类似，不要与AnonymousObservable弄混了。一个是观察者，一个是被观察者。
闭包的代码中根据事件类型调用 onNext 或 onError 或 onCompleted闭包，这个是最后一步的调用， 顺便提一下处理.error 和 .completed 事件时会调用disposable.dispose()，说明这两个事件发生后会取消订阅，回收资源，之后无法再发送序列了。
- 接着看
return Disposables.create( self.asObservable().subscribe(observer), disposable )
这里会创建Disposables，它传入两个disposeable，这里先不讲dispose，看第一个参数, 通过asObservable 调用subscribe，并把刚刚创建的observer传入，asObservable我们可以通过上面的类图看到它是ObservableConvertibleType的协议，Observable 必然实现了这个协议，实际上是返回自身。自身的类型其实是AnonymousObservable，所以就是通过self调用subscribe，subscribe在父类Producer实现了。 - 接着看Producer 的subscribe代码：
override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
if !CurrentThreadScheduler.isScheduleRequired {
// The returned disposable needs to release all references once it was disposed.
let disposer = SinkDisposer()
let sinkAndSubscription = self.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
}
}
}
image.png
所以最后会走到闭包中，关注下闭包的代码：调用了 self.run, 把参数observer 传进去， observer是外面创建的AnonymousObserver对象，我们整个流程只会有一个AnonymousObserver和一个AnonymousObservable其他地方看到的observer都是传进去的，所以看到observer简单想到是最开始创建的AnonymousObserver就行了。
- 现在到run方法了，AnonymousObservable实现了run方法：
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
- 这里比较关键：创建了AnonymousObservableSink对象，Sink翻译做“业务下沉”，表示这个类是管理者或者专门处理业务的。将observer和cancel传进去。observer还是通过调用链传来的。接着调用AnonymousObservableSink的run方法：sink.run(self)
- 看一下AnonymousObservableSink的run方法：
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
Parant 是 sink.run(self) 的self， self实际上是AnonymousObservable对象，就是唯一的被观察者，是用户通过create创建的。
- 所以接下来调用了AnonymousObservable对象的_subscribeHandler，同时把AnyObserver(self)最为参数传入，_subscribeHandler是个闭包，我们可以回到第1步看看。_subscribeHandler是最开始 create 被观察者传入的闭包，所以到这一步才开始执行第1步创建的闭包，这里还有个关键点：
AnyObserver(self), 创建了一个AnyObserver并把self作为参数传入，self是 AnonymousObservableSink对象啊，记住了后面会比较绕。 打开AnyObserver看看这个初始化方法：
/// Construct an instance whose `on(event)` calls `observer.on(event)`
///
/// - parameter observer: Observer that receives sequence events.
public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
self.observer = observer.on
}
是这个没问题，通过类图可以看到AnonymousObservableSink遵循了ObserverType，把on方法赋值给observer。
-
self.observer = observer.on, 把AnonymousObservableSink的on 方法复制给observer 这里的observer可不是AnonymousObserver对象，它是个闭包，类型是(Event<Element>) -> Void。诶，我是谁！我在哪里！😮💨，我们回到执行_subscribeHandler的地方。 - 第8步说到执行_subscribeHandler，也就是我们最开始create Observable的闭包：
let observable = Observable<Int>.create { (anyObserver) -> Disposable in
anyObserver.onNext(1)
anyObserver.onCompleted()
return Disposables.create()
}
- 当代码执行
anyObserver.onNext(1)， anyObserver是刚刚第8步创建的哦， AnyObserver没有找到方法，它其实在ObserverType extension中(AnyObserver 遵循了ObserverType)：
/// Convenience API extensions to provide alternate next, error, completed events
extension ObserverType {
/// Convenience method equivalent to `on(.next(element: Element))`
///
/// - parameter element: Next element to send to observer(s)
public func onNext(_ element: Element) {
self.on(.next(element))
}
执行了on方法
- AnyObserver有实现了 on 方法协议：
public func on(_ event: Event<Element>) {
return self.observer(event)
}
-
调用
self.observer(event), 这个observer 在第9步（self.observer = observer.on）由AnonymousObservableSink.on 赋值的闭包。所以解下来调用AnonymousObservableSink.on方法。 -
我们看看on方法实现：
func on(_ event: Event<Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
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()
}
}
}
我们看到AnonymousObservableSink 类的 run 方法和on 方法在整个流程中至关重要，run方法调用了订阅的handler，而on 方法处理了事件。这说明AnonymousObservableSink类处理主要业务逻辑，是整个流程的核心。
- 最终会调用
self.forwardOn(event), 这个forwardOn是在父类Sink定义的，然后跳到forwardOn看看：
final func forwardOn(_ event: Event<Observer.Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
- 然后调用
self._observer.on(event), 这个_observer 是在第6步创建AnonymousObservableSink对象时作为初始化参数赋值的。这个observer就是唯一的AnonymousObserver对象，还知道它是什么时候创建的吗？所以我们去AnonymousObserver找下on方法，没找到，去父类ObserverBase找到了：
case .next:
if load(self._isStopped) == 0 {
self.onCore(event)
}
- 接着调用onCore，在AnonymousObserver里：
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
终于绕出来了，调用self._eventHandler(event), _eventHandler 现在还记得是啥吗？啥时候被赋值的？_eventHandler是在第2步创建AnonymousObserver时被赋值的。所以接着调用第2步的代码：case .next(let value): onNext?(value) 调用onNext就是我们订阅时传入的闭包。最后打印1， 希望大家能看明白，下一篇文章会解析dispose流程。
网友评论