探索RxSwift的核心逻辑,我们从RxSwift的使用流程入手,带着自己的疑问去寻找答案,将会事半功倍。
RxSwift的使用
RxSwift的使用流程:创建序列、订阅信号、发送信号
// 1:创建序列
let ob = Observable<Any>.create { (obserber) -> Disposable in
// block_1 3:发送信号 block_1
obserber.onNext("RxSwift课程")
obserber.onCompleted()
return Disposables.create()
}
//2.订阅信号
let _ = ob.subscribe(onNext: { (text) in
//block_2
print("订阅到:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
上述代码,我们知道在创建序列时,后面的逃逸闭包block_1中包含发送信号的代码,在订阅信号时会创建闭包block_2,发送信号就是执行了闭包block_1,将RxSwift课程发送到闭包block_2。
那么在它内部在订阅序列是如何找到block_1的,而block_1是如何找到block_2的? 我们带着好奇,对RxSwift的核心逻辑进行分析。
创建序列
创建序列过程中做了什么？带着问题我们找到Create.swift下create的方法实现:
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
它是ObservableType类的扩展。
在代码create方法中,我们知道创建序列实际上创建的是AnonymousObservable对象,从字面上理解,这是一个匿名可观察对象.
下图是AnonymousObservable类的继承关系。
final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable
let _subscribeHandler: SubscribeHandler
init(_ subscribeHandler: @escaping SubscribeHandler) {
self._subscribeHandler = subscribeHandler
}
}
我们知道AnonymousObservable继承自Producer, 拥有Producer的subscribe方法,并且将创建序列时传入的逃逸闭包block_1保存至_subscribeHandler
总结
创建序列主要做了如下操作:
1.创建并返回AnonymousObservable
2.AnonymousObservable继承自Producer, 拥有Producer的subscribe方法
3.将创建序列时传入的逃逸闭包block_1保存至_subscribeHandler
订阅信号
下列是订阅信号subscribe的源码,它是ObservableType的扩展
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观察者对象,初始化的参数是一个逃逸闭包，保存了外界的 onNext,onError , onCompleted , onDisposed 的处理回调闭包。
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self._eventHandler = eventHandler
}
}
AnonymousObserver的初始化方法中,将闭包保存在_eventHandler
以下是AnonymousObserver的继承关系:
最后在Disposables.create中,执行self.asObservable().subscribe(observer),asObservable()是RxSwift保持一致性的写法,self.asObservable().subscribe(observer)等同于self.subscribe(observer)。在AnonymousObservablea并没有找到subscribe方法, 通过AnonymousObservable继承关系,我们在Producer找到了subscribe的实现。
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
}
}
}
代码中let sinkAndSubscription = self.run(observer, cancel: disposer)得知,Producer又将run的实现分发到具体的子类:
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)
}
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
创建AnonymousObservableSink对象：
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
}
AnonymousObservableSink继承自Sink
class Sink<O : ObserverType> : Disposable {
init(observer: O, cancel: Cancelable) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self._observer = observer
self._cancel = cancel
}
}
将观察者observer保存至Sink的_observer
let subscription = sink.run(self)中，sink.run主要是将依赖下沉,使得分工更加明确。
进入AnonymousObservableSink的run方法:
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
这里的parent正是我们之前创建的AnonymousObservable序列, parent._subscribeHandler这里执行的正是我们创建序列时保存的_subscribeHandler,就是闭包block_1，到这里，我们已经知道订阅序列的时候，实际上会去执行创建时保存的逃逸闭包block_1。
传递的参数是AnyObserver(self),这个AnyObserver是什么?
public struct AnyObserver<Element> : ObserverType {
/// 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
}
public init<O : ObserverType>(_ observer: O) where O.E == Element {
self.observer = observer.on
}
}
AnyObserver是一个结构体,AnyObserver(self)保存了AnonymousObservableSink的on函数。注意这个地方有2个init的方法,调用的是下面这个,所以保存的是AnonymousObservableSink的on函数，而不是AnonymousObservableSink本身。
发送信号
obserber.onNext("RxSwift课程")
从订阅信号分析的结果,我们知道闭包block_1的参数实际上是AnyObserver类型，obserber.onNext等同于调用AnyObserver.onNext,AnyObserver类中并不存在onNext方法，因为我们顺着继承关系在ObserverType找到了onNext
extension ObserverType {
/// Convenience method equivalent to `on(.next(element: E))`
///
/// - parameter element: Next element to send to observer(s)
public func onNext(_ element: E) {
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方法中调用self.on(.next(element)), 等同于调用了AnyObserver.on(.next(element))方法
public func on(_ event: Event<Element>) {
return self.observer(event)
}
AnyObserver.on方法中self.observer保存的是AnonymousObservableSink.on
参数为.next函数, .next函数的参数正式我们最终的参数RxSwift课程.
我们来理一下整个流程:
obserber.onNext("RxSwift课程") ---> AnyObserver.on(.next("RxSwift课程"))
--->AnonymousObservableSink.on(.next("RxSwift课程"))
这里obserber.onNext的调用又回到AnonymousObservableSink类中,
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
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()
}
}
}
}
我们看到在AnonymousObservableSink.on方法中,调用了self.forwardOn(event)方法,
AnonymousObservableSink继承自Sink。在Sink下找到forwardOn方法:
class Sink<O : ObserverType> : Disposable {
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)
}
}
forwardOn方法中调用self._observer.on(event), self._observer就是订阅时保存的观察者AnonymousObserver,最终得出obserber.onNext("RxSwift课程")实际上是AnonymousObserver.on(.next("RxSwift课程")).
通过集成关系，在ObserverBase类中找到了AnonymousObserver.on的实现：
class ObserverBase<ElementType> : Disposable, ObserverType {
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)
}
}
}
}
代码中on的实现调用了self.onCore(event),将具体业务实现分发到子类AnonymousObserver
final class AnonymousObserver<ElementType> : ObserverBase<ElementType> {
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
}
AnonymousObserver的onCore调用self._eventHandler(event),_eventHandler正是订阅序列时AnonymousObserver的初始化方法中保存的_eventHandler
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()
}
}
通过switch event 判断事件类型,并最终调用 onNext?(value).然后就是外界onNext闭包的调用。
总结:
1.RxSwift万物皆序列,通过asObservable保持一致性。
2.使用Sink，将依赖下沉，使代码更加独立,逻辑更加清晰。
3.一次订阅,持续响应。
