RxSwift5.0核心逻辑理解
2020-02-05 本文已影响0人
数字d
Demo代码实现内容
//1.创建序列
let ob = Observable<Any>.create { (obserber) -> Disposable in
// 3:发送信号
obserber.onNext("框架")
obserber.onCompleted()
// obserber.onError(NSError.init(domain: "coocieeror", code: 10087, userInfo: nil))
return Disposables.create()
}
// 2:订阅信号
// AnonymousObserver - event .next -> onNext()
// _eventHandler
// AnonymousObservable._subscribeHandler(observer)
// 销毁
let _ = ob.subscribe(onNext: { (text) in
//4.订阅到
print("订阅到:\(text)")
}, onError: { (error) in
print("error: \(error)")
}, onCompleted: {
print("完成")
}) {
print("销毁")
}
序列继承链
(class)Observable遵循了(protocol)ObservableType，在Create.swift文件中ObservableType通过扩展(extension)添加了create方法
extension ObservableType {
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
}
在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
}
override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
//管道，可以理解为一个Manager
let sink = AnonymousObservableSink(observer: observer, cancel: cancel)
let subscription = sink.run(self)
return (sink: sink, subscription: subscription)
}
}
AnonymousObservable是一个类（class），这个匿名序列继承自Producer
class Producer<Element> : Observable<Element> {
override init() {
super.init()
}
//重写父类的subscribe方法，订阅事件
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
...
}
func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
rxAbstractMethod()
}
}
Producer继承于Observable，Observable是可观察序列，遵循ObservableType协议，是所有序列的基类,ObservableType协议定义了一个函数subscribe，所有遵循这个协议的序列都具有subscribe的功能。
public class Observable<Element> : ObservableType {
public typealias E = Element
init() {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()//RxSwift自己的引用计数，创建就会增加，销毁就会减少
#endif
}
public func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
rxAbstractMethod()//抽象方法，没有实现，等着被重写
}
public func asObservable() -> Observable<E> {
return self
}
deinit {
#if TRACE_RESOURCES
_ = Resources.decrementTotal()//根据这个可以知道是否存在内存问题
#endif
}
internal func composeMap<R>(_ transform: @escaping (Element) throws -> R) -> Observable<R> {
return _map(source: self, transform: transform)
}
}
Observable类本来就是一个可观察序列，为什么突然又出现一个asObservable函数呢？
为了统一管理，使得所有的类型都是序列类。
比如：UISwitch().rx.value，虽然UISwitch也可以.出rx.出value，但是查看value的底层遵循了ObservableType协议，也可以订阅事件，但是它始终不是Observable对象，为了规范统进行转化。“as”类似于将类型强转。
public var value: RxCocoa.ControlProperty<Bool> { get } //是ControlProperty类型的
public struct ControlProperty<PropertyType> : ControlPropertyType { //遵循了ControlPropertyType协议
public typealias E = PropertyType
...
}
public protocol ControlPropertyType : ObservableType, ObserverType { //遵循了ObservableType协议
/// - returns: `ControlProperty` interface
func asControlProperty() -> ControlProperty<E>
}
序列Observable提供所有功能，有asObservable（来源于协议），提供subscribe接口，使得所有继承于Observable的序列都具有订阅的功能。
2.png
订阅者继承链和订阅流程
Observable在subscribe订阅时会创建一个observer订阅者
public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
-> Disposable {
let disposable: Disposable
...
创建一匿名订阅者，类型E是调用者传入的序列类型
let observer = AnonymousObserver<E> { event in
...
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<ElementType> : ObserverBase<ElementType> {
typealias Element = ElementType
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>) {
return self._eventHandler(event)
}
#if TRACE_RESOURCES
deinit {
_ = Resources.decrementTotal()
}
#endif
}
AnonymousObserver继承于ObserverBase，ObserverBase的源码如下：
class ObserverBase<ElementType> : Disposable, ObserverType {
typealias E = ElementType
private let _isStopped = AtomicInt(0)
func on(_ event: Event<E>) { //实现ObserverType协议方法，Event是枚举类型
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<E>) {
rxAbstractMethod()
}
func dispose() {
fetchOr(self._isStopped, 1)
}
}
Event源码：
public enum Event<Element> {
/// Next element is produced.
case next(Element)
/// Sequence terminated with an error.
case error(Swift.Error)
/// Sequence completed successfully.
case completed
}
ObserverBase具有不断发送信号的功能.
ObserverType是一个协议，源码如下
public protocol ObserverType {
/// The type of elements in sequence that observer can observe.
associatedtype E //关联绑定类型
/// Notify observer about sequence event.
///
/// - parameter event: Event that occurred.
func on(_ event: Event<E>) //协议方法
}
Producer.subscribe中的“self.asObservable().subscribe(observer)”到底由谁来进行调用呢？
不知道它是什么类型，但是为了统一，所以调用了asObservable，调用subscribe方法。AnonymousObservable里面没有subscribe方法，那么就开始去查找他的父类Producer，Producer继承了Observable，并且实现了subscribe这个抽象类的实现。
class Producer<Element> : Observable<Element> {
override init() {
super.init()
}
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
}
}
}
func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
rxAbstractMethod()
}
}
Producer统一处理避免子类的下发产生冗余代码，且并不影响基类，基类也不需要关心具体的实现。具体的run方法实现应该由子类单独去处理，体现出RxSwift的分工非常的明确。这里使用的AnonymousObersevable，所以具体的run实现源码在AnonymousObersevable里，源码如下：
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)
}
}
4.png
核心重点
AnonymousObservable会调用父类Producer的subscribe函数，subscribe函数内，会调用AnonymousObservable具体的run方法，Create.swift文件中AnonymousObservable的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)
}
AnonymousObservableSink继承于Sink，Sink中就有observer和cancel这两个属性，在此进行保存。
class Sink<O : ObserverType> : Disposable {
fileprivate let _observer: O
fileprivate let _cancel: Cancelable
fileprivate let _disposed = AtomicInt(0)
...
init(observer: O, cancel: Cancelable) {
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
self._observer = observer
self._cancel = cancel
}
...
}
run函数中，sink调用了本类中的AnonymousObservable.run
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
typealias Parent = AnonymousObservable<E>
因此，AnonymousObservableSink这个管子拥有了 订阅者、销毁者、序列、调度环境，处理的就是具体的业务逻辑。
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
// 外界传进来的闭包，传递给AnonymousObservable
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
}
}
这里的parent就是我们的序列AnonymousObservable，调用_subscribeHandler函数就是我们一开始creat序列的时候外界传入的闭包。
run函数中_subscribeHandler(AnyObserver(self))中的AnyObserver(self)是什么呢？是不是AnonymousObserver呢？
AnyObserver是一个结构体，相当于一个序列化器，源码如下：
public struct AnyObserver<Element> : ObserverType {
public typealias EventHandler = (Event<Element>) -> Void
private let observer: EventHandler
public init(eventHandler: @escaping EventHandler) {
self.observer = eventHandler
}
public init<O : ObserverType>(_ observer: O) where O.E == Element {
// 这里的observer就是我们传进来的sink，当前的observer是一个函数
self.observer = observer.on
}
public func on(_ event: Event<Element>) {
return self.observer(event)
}
public func asObserver() -> AnyObserver<E> {
return self
}
}
observer.on就是调用了管子的on函数，核心逻辑就在AnonymousObservableSink里面，源码：
final private class AnonymousObservableSink<O: ObserverType>: Sink<O>, ObserverType {
typealias E = O.E
typealias Parent = AnonymousObservable<E>
// state
private let _isStopped = AtomicInt(0)
#if DEBUG
fileprivate let _synchronizationTracker = SynchronizationTracker()
#endif
override init(observer: O, cancel: Cancelable) {
super.init(observer: observer, cancel: cancel)
}
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()
}
}
}
func run(_ parent: Parent) -> Disposable {
return parent._subscribeHandler(AnyObserver(self))
}
}
AnyObserver这个结构体遵循了ObserverType协议，因此它也有onNext这个方法，而onNext这个方法就是调用了on函数，而这个on函数也就是调用的传进来的管子AnonymousObservableSink的on函数块，订阅者已经加载完毕。
extension ObserverType {
public func onNext(_ element: E) {
self.on(.next(element))
}
public func onCompleted() {
self.on(.completed)
}
public func onError(_ error: Swift.Error) {
self.on(.error(error))
}
}
管子中的self.forwardOn(event)函数调用了父类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)
}
forwardOn函数中的_observer就是从订阅的时候Producer调用subscribe传过来的那个AnonymousObserver。
RxSwiftPic.png
源码地址：https://gitee.com/xgkp/RX2.0login.git
源码解读图片地址：https://gitee.com/xgkp/Rx_5.0.git
