前言：这几篇关于RxSwift源码分析的文章主要是对源码进行解析，不涉及到RxSwift的具体使用。具体使用可以查看RxSwift中文文档或者github。
在RxSwift的使用中，我们需要先创建序列，然后订阅信号、发送信号来实现一个完整的流程。先举一个简单的例子：
// 1:创建序列
let observable = Observable<Any>.create { (obserber) -> Disposable in
// 3:发送信号
obserber.onNext("发送信号")
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("销毁")
}
我们需要重点关注的是：
- 第一行代码中的create的闭包什么时候执行
- 订阅信号subscribe的闭包什么时候执行
- 信号从发送出去到接收流程是怎样的
下面针对这几个问题来一一探索
第一步create函数
首先来看看create函数，它返回一个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函数
源码：
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
)
}
从上面的代码可以看出，首先创建了一个AnonymousObserver对象，在初始化时传递了一个闭包作为参数并且保存下来self._eventHandler = eventHandler。
AnonymousObserver是匿名观察者,用于存储和处理事件的闭包。
然后在最后有self.asObservable().subscribe(observer)这样一行代码，asObservable返回的是对象本身，然后调用subscribe这个函数并且把创建的AnonymousObserver对象传递过去，会来到AnonymousObservable这个类里面，但是发现这个类里面没有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作为参数传过来。
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方法中，先创建一个AnonymousObservableSink对象并持有observer，然后调用这个对象的run方法把self传递过去，也就是把observable作为参数。
AnonymousObservableSink这个类将可观察者Observable和观察者Observer链接起来,实现事件的传递，起到一个桥梁的作用。
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
在这里触发了_subscribeHandler的调用，这里的_subscribeHandler就是之前create函数的闭包。到这里就解决了第一个问题“第一行代码中的create的闭包什么时候执行”。
在这个方法中把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
}
}
可以看到在构造函数中有一行代码self.observer = observer.on，就是把AnonymousObservableSink类的on函数赋值给AnyObserver类的observer变量。从这里就可以明白为什么这行代码observer.onNext("发送信号") 最终会触发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函数里面创建的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函数传入的闭包。
到这里就可以理解第二个和第三个问题了。
总结
从整个流程来看，我们可以体会到函数响应式编程的魅力，代码简介，思路非常的清晰。
梳理一下整个流程：
tips：由于在RxSwift中广泛的使用了协议和泛型，还有很多抽象方法和同名方法，导致在刚开始阅读源码的时候可能有些不是很习惯，再加上Xcode跟踪代码不是很强大，阅读起来有一些障碍。所以在阅读源码的时候需要耐心一点，不要找错了地方，可以找一些重点代码阅读，不要盲目的从头看到尾。
有问题或者建议和意见，欢迎大家评论或者私信。
喜欢的朋友可以点下关注和喜欢，后续会持续更新文章。
