前言：這幾篇關于RxSwift源碼分析的文章主要是對源碼進行解析业踏，不涉及到RxSwift的具體使用鞍陨。具體使用可以查看RxSwift中文文檔或者github。
在RxSwift的使用中，我們需要先創(chuàng)建序列乳乌，然后訂閱信號、發(fā)送信號來實現(xiàn)一個完整的流程市咆。先舉一個簡單的例子：
// 1:創(chuàng)建序列
let observable = Observable<Any>.create { (obserber) -> Disposable in
// 3:發(fā)送信號
obserber.onNext("發(fā)送信號")
obserber.onCompleted()
// obserber.onError(NSError.init(domain: "fail", code: 10087, userInfo: nil))
return Disposables.create()
}
// 2:訂閱信號
let _ = observable.subscribe(onNext: { (text) in
print("訂閱到信號:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("銷毀")
}
我們需要重點關注的是：
1汉操、第一行代碼中的create的閉包什么時候執(zhí)行
2、訂閱信號subscribe的閉包什么時候執(zhí)行
3蒙兰、信號從發(fā)送出去到接收流程是怎樣的
下面針對這幾個問題來一一探索
第一步create函數(shù)
首先來看看create函數(shù)磷瘤，它返回一個AnonymousObservable對象，AnonymousObservable是匿名可觀察者,用來存儲產生事件的閉包（self._subscribeHandler = subscribeHandler）和激活處理事件閉包的入口（run方法）搜变。
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
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
}
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是Observable的子類采缚，它們的繼承關系是：AnonymousObservable -> Producer -> Observable -> ObservableType -> ObservableConvertibleType
第二步subscribe函數(shù)
源碼：
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
let disposable: Disposable
if let disposed = onDisposed {
disposable = Disposables.create(with: disposed)
}
else {
disposable = Disposables.create()
}
#if DEBUG
let synchronizationTracker = SynchronizationTracker()
#endif
let callStack = Hooks.recordCallStackOnError ? Hooks.customCaptureSubscriptionCallstack() : []
let observer = AnonymousObserver<E> { 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
)
}
從上面的代碼可以看出，首先創(chuàng)建了一個AnonymousObserver對象挠他，在初始化時傳遞了一個閉包作為參數(shù)并且保存下來self._eventHandler = eventHandler扳抽。
AnonymousObserver是匿名觀察者,用于存儲和處理事件的閉包。
然后在最后有self.asObservable().subscribe(observer)這樣一行代碼殖侵，asObservable返回的是對象本身贸呢，然后調用subscribe這個函數(shù)并且把創(chuàng)建的AnonymousObserver對象傳遞過去，會來到AnonymousObservable這個類里面拢军，但是發(fā)現(xiàn)這個類里面沒有subscribe方法楞陷，我們往父類Producer里面找到這個方法，源碼如下：
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 = 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
}
}
}
在這里調用了run方法茉唉，也就是AnonymousObservable這個類里面的run方法固蛾，把observer作為參數(shù)傳過來结执。
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)
}
在run方法中，先創(chuàng)建一個AnonymousObservableSink對象并持有observer艾凯，然后調用這個對象的run方法把self傳遞過去昌犹，也就是把observable作為參數(shù)。
AnonymousObservableSink這個類將可觀察者Observable和觀察者Observer鏈接起來,實現(xiàn)事件的傳遞览芳，起到一個橋梁的作用斜姥。
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
在這里觸發(fā)了_subscribeHandler的調用，這里的_subscribeHandler就是之前create函數(shù)的閉包沧竟。到這里就解決了第一個問題“第一行代碼中的create的閉包什么時候執(zhí)行”铸敏。
在這個方法中把self轉換成AnyObserver對象，也就是把AnonymousObservableSink對象轉換成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
}
}
可以看到在構造函數(shù)中有一行代碼self.observer = observer.on糕非，就是把AnonymousObservableSink類的on函數(shù)賦值給AnyObserver類的observer變量蒙具。從這里就可以明白為什么這行代碼observer.onNext("發(fā)送信號") 最終會觸發(fā)AnonymousObservableSink.on事件。
// AnonymousObservableSink類on方法
func on(_ event: Event<E>) {
#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()
}
}
}
然后進入到Sink類的forwardOn方法
final func forwardOn(_ event: Event<O.E>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
if isFlagSet(self._disposed, 1) {
return
}
self._observer.on(event)
}
這里的_observer就是第二步調用subscribe函數(shù)里面創(chuàng)建的observer對象朽肥。
會先進入到父類的ObserverBase的on方法
func on(_ event: Event<E>) {
switch event {
case .next:
if load(self._isStopped) == 0 {
self.onCore(event)
}
case .error, .completed:
if fetchOr(self._isStopped, 1) == 0 {
self.onCore(event)
}
}
}
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
這里的_eventHandler閉包就是第二步調用subscribe函數(shù)傳入的閉包禁筏。
到這里就可以理解第二個和第三個問題了。
總結
從整個流程來看衡招，我們可以體會到函數(shù)響應式編程的魅力篱昔，代碼簡介，思路非常的清晰始腾。
梳理一下整個流程：
tips：由于在RxSwift中廣泛的使用了協(xié)議和泛型州刽，還有很多抽象方法和同名方法，導致在剛開始閱讀源碼的時候可能有些不是很習慣浪箭，再加上Xcode跟蹤代碼不是很強大穗椅，閱讀起來有一些障礙。所以在閱讀源碼的時候需要耐心一點奶栖，不要找錯了地方匹表，可以找一些重點代碼閱讀，不要盲目的從頭看到尾驼抹。
有問題或者建議和意見桑孩，歡迎大家評論或者私信。
喜歡的朋友可以點下關注和喜歡框冀，后續(xù)會持續(xù)更新文章流椒。
