RxSwift-核心逻辑分析

基本名词解析

  • Observable:可监听序列,他是事件的生产者。举个例子,你可以将温度看作是一个序列,然后监测这个温度值,最后对这个值做出响应。例如:当室温高于 33 度时,打开空调降温。
  • Observer:观察者,是用来监听事件,然后它需要这个事件做出响应。比如监听当前室温,室温高于 33 度时,打开空调降温。打开空调降温就是观察者

RxSwift核心逻辑分析

RxSwift中,对事件的处理遵循以下三个步骤:

  1. 创建可监听序列Observable
  2. 订阅序列,创建观察者Observer对此序列进行观察。
  3. 序列中产生事件,观察者对序列中的事件进行响应。

创建序列流程分析

下面这段代码,是RxSwift的最基本使用,我们就此开始进行分析

let ob = Observable<Int>.create { (observer) -> Disposable in
observer.onNext(1)
return Disposables.create()
}

ob.subscribe { (value) in
print(value)
} onError: { (error) in
print(error)
} onCompleted: {
print("完成")
} onDisposed: {
print("销毁")
}
.disposed(by: disposeBag)

Observable是什么呢?我们点击进去发现,它就是一个抽象类,遵循了ObservableType协议。

public class Observable<Element> : ObservableType {
// ...省略实现
}

点击create方法,发现它是ObservableType协议的方法, 通过全局搜索,可以发现它的实现在Create.swift文件中。

extension ObservableType {
public static func create(_ subscribe: @escaping (AnyObserver<Element>) -> Disposable) -> Observable<Element> {
AnonymousObservable(subscribe)
}
}

AnonymousObservable是一个内部类, 它有一个subscribeHandler属性,存储我们刚刚创建的闭包,也就是红框中的代码。

final private class AnonymousObservable<Element>: Producer<Element> {
typealias SubscribeHandler = (AnyObserver<Element>) -> Disposable

let subscribeHandler: SubscribeHandler

init(_ subscribeHandler: @escaping SubscribeHandler) {
self.subscribeHandler = subscribeHandler
}

...
}

序列的创建类图和方法调用流程如下:

订阅流程分析

序列创建完毕以后,我们还需要对序列进行订阅,才能观察序列中的事件,然后做出响应。ob.subscribe(onNext:onError:onCompleted:)的实现如下:

public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
...

// 省略无关代码,保留关键代码
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

final class AnonymousObserver<Element>: ObserverBase<Element> {
typealias EventHandler = (Event<Element>) -> Void

private let eventHandler : EventHandler

init(_ eventHandler: @escaping EventHandler) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self.eventHandler = eventHandler
}

override func onCore(_ event: Event<Element>) {
self.eventHandler(event)
}

#if TRACE_RESOURCES
deinit {
_ = Resources.decrementTotal()
}
#endif
}

可以发现,AnonymousObserver也存在一个属性eventHandler,用来存储事件的响应处理,根据事件的不同,调用不同的响应处理。例如,观察到的事件是.next,则调用我们的onNext处理print(value),如下图所示。

可是,到现在为止,上面的这段代码,都只是调用AnonymousObserver的初始化方法,把闭包进行存储,我们还无法知道序列和事件响应直接是如何进行关联的。

return位置,我们可以看到调用了一个subscribe方法

return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)

调用了Observablesubscribe(_:)方法,这是个抽象方法,具体实现在它的子类Producer

class Producer<Element>: Observable<Element> {
override init() {
super.init()
}

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
}
}
}

func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
rxAbstractMethod()
}
}

subscribe最终调用的是run(_:cancel:)方法,Producerrun(_:cancel:)方法也是一个抽象方法,得看它的子类AnonymousObservable

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)
}

最终调用的是AnonymousObservableSinkrun(_:)方法。

func run(_ parent: Parent) -> Disposable {
parent.subscribeHandler(AnyObserver(self))
}

关联的地方来咯

  • parent是我们传进去的AnonymousObservable对象,它的属性subscribeHandler保存着最开始创建序列的闭包。
  • parent.subscribeHandler(AnyObserver(self))对闭包进行了调用,这样我们就完成了对序列事件信号的监听。

整体流程如下:

响应流程分析

通过上面的步骤,我们监听对序列进行了监听,那我们是怎么把事件发送去的呢?在闭包中,我们有这样一段代码:

let ob = Observable<Int>.create { (observer) -> Disposable in
observer.onNext(1)
return Disposables.create()
}

这里的observer是一个AnyObserver对象,在parent.subscribeHandler(AnyObserver(self))调用的时候进行的初始化,此时的selfAnonymousObservableSink对象。

public struct AnyObserver<Element> : ObserverType {

public typealias EventHandler = (Event<Element>) -> Void

// 此时的observer存储的是AnonymousObservableSink的on方法
private let observer: EventHandler

public init(eventHandler: @escaping EventHandler) {
self.observer = eventHandler
}
// 走的是这个初始化方法
public init<Observer: ObserverType>(_ observer: Observer) where Observer.Element == Element {
self.observer = observer.on
}

public func on(_ event: Event<Element>) {
self.observer(event)
}

public func asObserver() -> AnyObserver<Element> {
self
}
}

AnyObserver是一个结构体,可以发现,它并没有onNext方法,那么只可能是ObserverType协议的方法。

public protocol ObserverType {

associatedtype Element

func on(_ event: Event<Element>)
}

extension ObserverType {

public func onNext(_ element: Element) {
self.on(.next(element))
}

public func onCompleted() {
self.on(.completed)
}

public func onError(_ error: Swift.Error) {
self.on(.error(error))
}
}

onNext(_:)方法调用了on(_:)方法,AnyObserver.on(_:)调用了刚刚保存的AnonymousObservableSink.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()
}
}
}

最后调用了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)
}

AnonymousObservableSink初始化的时候,使用属性observer保存了AnonymousObserver对象。在其父类中找到on(_:)方法。

class ObserverBase<Element> : Disposable, ObserverType {
private let isStopped = AtomicInt(0)

func on(_ event: Event<Element>) {
switch event {
case .next:
if load(self.isStopped) == 0 {
self.onCore(event)
}
case .error, .completed:
if fetchOr(self.isStopped, 1) == 0 {
self.onCore(event)
}
}
}

func onCore(_ event: Event<Element>) {
rxAbstractMethod()
}

func dispose() {
fetchOr(self.isStopped, 1)
}
}

最终调用AnonymousObserveronCore(_:)方法。

override func onCore(_ event: Event<Element>) {
self.eventHandler(event)
}

这样,就调用了响应处理的闭包,也就是subscribeonNextonErroronCompleted

总结

整体类图和流程可以总结为下图