RxSwift學習--核心邏輯再探
前言
經過上一篇內容RxSwift學習--核心邏輯初探,對RxSwift有了些初步的認知，下面經過源碼來看一下RxSwift到底有多騷html
RxSwift核心邏輯再探
先把上篇中的例子代碼搬過來：react
//第一步:建立序列
//在create()函數中傳入一個閉包，任務是對每個過來的訂閱進行處理
let ob = Observable<Any>.create { (observer) -> Disposable in
// 第三步:發送信號（onCompleted和onError只能發送一個）
observer.onNext("你好啊")
observer.onCompleted()
// observer.onError(NSError.init(domain: "loser", code: 10010, userInfo: nil))
return Disposables.create()
//第二步:訂閱信息
//當咱們訂閱了Observable的消息後，只要Observable的事件觸發，都會經過onNext這個閉包告訴咱們。
let _ = ob.subscribe(onNext: { (text) in
print("訂閱到:\(text)") //這裏會監聽到訂閱的Observable事件
}, onError: { (error) in
print("error: \(error)") //當發生錯誤時，會回調這裏
}, onCompleted: { // 當序列執行完畢時，會回調這裏。
print("完成")
}) {
print("銷燬")
}
複製代碼
下面咱們根據代碼進行具體分析：swift
1.建立序列Observable<Any>.create()
在執行let ob = Observable<Any>.create { (obserber) -> Disposable in } 這句代碼時，是建立了一個可觀察序列，點進create()方法的源碼api
extension ObservableType {
/*
Creates an observable sequence from a specified subscribe method implementation.
- seealso: [create operator on reactivex.io](http://reactivex.io/documentation/operators/create.html)
- parameter subscribe: Implementation of the resulting observable sequence's `subscribe` method. - returns: The observable sequence with the specified implementation for the `subscribe` method. */ public static func create(_ subscribe: @escaping (RxSwift.AnyObserver<Self.E>) -> Disposable) -> RxSwift.Observable<Self.E> } 複製代碼
根據註釋路由可知create()方法是在Create.swift文件中實現的bash
public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
return AnonymousObservable(subscribe)
}
複製代碼
在create()方法中返回了一個匿名內部類---AnonymousObservable(匿名序列)，用來存儲產生事件的閉包（self._subscribeHandler = subscribeHandler）和激活處理事件閉包的入口（run方法）在跟進去這個類，閉包
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)
}
}
複製代碼
能夠看到在匿名序列的裏面是把傳給create()方法的閉包（也就是序列訂閱的回調,self._subscribeHandler = subscribeHandler）保存下來了dom
2.序列訂閱ob.subscribe(onNext:)
一樣的，跟進去訂閱信號的這個方法的源碼， 在這裏跟源碼，跟進去的是ObserveableType.subscribe這個方法 ide
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
)
}
複製代碼
能夠看到在調用ob.subscribe()這個方法的時候，onNext, onError,onComplete, onDisposed這四個閉包都是做爲參數傳遞到這個訂閱函數中來， 在這段代碼裏一樣的也建立了一個匿名觀察者let observer = AnonymousObserver<E>用於存儲和處理事件的閉包，而對於這個觀察者會傳入一個帶有event的閉包，在前面已經知道event是一個枚舉類型，有三種事件類型.next，.error，.completed,因此只要這個觀察者調用了這三種事件，那麼相應的就會調用onNext，onError，onCompleted這三個方法的具體實現，但是這裏的observer觀察者只是一個局部變量，它怎麼和外界的觀察者聯繫起來呢？函數
騷就騷在這個訂閱方法的return這句代碼：post
return Disposables.create(
self.asObservable().subscribe(observer),
disposable
)
複製代碼
其實這個self.asObservable()就是咱們在最開始定義的序列let ob = Observable<Any>,而subscribe()就是回調了咱們剛剛建立的observer閉包，而observer就會調用傳入的監聽序列消息閉包onNext, onError,onCompleted, 那麼這個subscribe()是怎麼回調了observer呢？
可知self.asObservable()是AnonymousObservable類型的，可是經過前面的AnonymousObservable類的源碼能夠發現，它並無subscribe()這個方法，可是AnonymousObservable是繼承於Producer(生產者)的，下面來看一下Producer這個類的源碼：
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()
}
}
複製代碼
在這裏面subscribe()方法會把傳入的observer，調用self.run(observer, cancel: disposer);那麼再次跟進去看源碼；至於這個run()方法具體怎麼調用，應該仍是交給Producer的子類AnonymousObservable去實現的，那麼跟進去，代碼再次回到了AnonymousObservable類裏面：
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對象並持有observer，而後這個AnonymousObservableSink類會調用它本身的run()方法,並傳入self,再次跟進去sink.run(self),發現來到了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))
}
}
複製代碼
能夠看到AnonymousObservableSink的run()方法，返回parent._subscribeHandler(),而這個Parent,在這個AnonymousObservableSink類中已經取了別名，那就是typealias Parent = AnonymousObservable<E>;在前面咱們已經知道，AnonymousObservable匿名序列的裏面是把傳給create()函數的閉包保存了下來，那麼return parent._subscribeHandler(AnyObserver(self))這個_subscribeHandler就是以前create()函數的閉包,在這個_subscribeHandler(AnyObserver(self))方法中把self轉換成AnyObserver對象，也就是把AnonymousObservableSink對象轉換成AnyObserver對象.
如今接着來看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
}
}
複製代碼
經過剛剛的AnyObserver(self)咱們應該知道這裏的self指的是AnonymousObservableSink,因此在AnyObserver的源碼中，能夠看到在構造函數中有一行代碼self.observer = observer.on,就是把AnonymousObservableSink類的on函數賦值給AnyObserver的observer變量.注意這裏保存的是AnonymousObservableSink.on.
看到這裏是否能明白，在調用ob.subscribe方法的時候，建立了一個AnonymousObserver對象，並把閉包傳入，可是這個對象最終走向的倒是繼承於Producer的AnonymousObservable類的subscribe()方法,通過調用self.run()以後，最後返回AnonymousObservable._subscribeHandler()方法，也就是ob.create()方法。這部分代碼先分析到這裏，下面來分析下發送信號的代碼。
3.發送信號observer.onNext("你好啊")
在剛開始分析create()方法的時候，就已經知道，observer.onNext("你好啊")中的observer是AnyObserver類型的，但是在上面的AnyObserver源碼中咱們能夠清楚地看到它並無onNext()方法，那就去它的父類ObserverType看一下，
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))
}
}
複製代碼
在這裏能夠看到調用ObserverType的onNext方法，返回ObserverType的on(.next(element)),因此在observer.onNext("你好啊")調用onNext()方法，其實調用的就是observer.on(.next()),在前面的AnyObserver的源碼中，已經知道AnyObserver的observer變量保存的是AnonymousObservableSink.on，一樣的看一下在AnyObserver的源碼中的on()方法
public func on(_ event: Event<Element>) {
return self.observer(event)
}
複製代碼
這裏返回的是self.observer(event),也就是說observer.on(.next())能夠是self.observer(.next())，然而self.observer = AnonymousObservableSink.on,因此self.observer(.next())能夠是AnonymousObservableSink.on(.next()),這裏又回到了AnonymousObservableSink這個類了，（AnonymousObservableSink的源碼在前面已經出現了）這是饒了一圈又回來了，下面把AnonymousObservableSink的on(event)方法單獨拿出來：
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()
}
}
}
複製代碼
在on(event)方法中有句重要的代碼self.forwardOn(event),再跟進去self.forwardOn(event)這個方法，能夠看到進入到Sink類的方法中，這裏AnonymousObservableSink繼承於Sink：
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)
}
複製代碼
在這個方法中self._observer.on(event)代碼中，這個self._observer就是在初始化AnonymousObservableSink時候傳入的observer( let sink = AnonymousObservableSink(observer: observer, cancel: cancel))，然而這個observer是在subscribe()訂閱方法中由let observer = AnonymousObserver建立的observer,那麼等因而這裏又來到了AnonymousObserver中,那麼下面就來看一下調用AnonymousObserver.on(event)會發生什麼，如今來看一下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類中並無on方法，可是AnonymousObserver是繼承於ObserverBase的，在ObserverBase的源碼中咱們能夠看到：
class ObserverBase<ElementType> : Disposable, ObserverType {
typealias E = ElementType
private let _isStopped = AtomicInt(0)
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)
}
}
}
func onCore(_ event: Event<E>) {
rxAbstractMethod()
}
func dispose() {
fetchOr(self._isStopped, 1)
}
}
複製代碼
調用ObserverBase的on()方法會返回self.onCore(event);這裏回到子類AnonymousObserver去執行這個方法；
override func onCore(_ event: Event<Element>) {
return self._eventHandler(event)
}
複製代碼
能夠看到這裏返回的是self._eventHandler(event);也就是AnonymousObserver._eventHandler(event)經過源碼可知這裏的AnonymousObserver._eventHandler(event)就是傳入的eventHandler(event),也就是序列訂閱方法ob.subscribe()傳入的閉包，那麼就來看下在subscribe()訂閱方法中AnonymousObserver建立的observer到底對這個eventHandler(event)作了什麼事情：
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()
}
}
複製代碼
由於Event是枚舉值這裏會根據observer傳入的event來判斷到底該響應onNext(),onError(),onCompleted()的具體方法，這樣就是把create()方法中的observer.onNext("你好啊")與subscribe()訂閱方法中onNext()聯繫起來了。這樣就說明了爲何觀察者調用了onNext()方法，序列可以訂閱到這個方法裏的內容。
總結
囉囉嗦嗦寫了這麼多，算是對RxSwift的核心邏輯有了個初步的認識，不等不說RxSwift很強大，足夠的"騷氣"。目前寫到這裏，對RxSwift僅僅只是略知皮毛，冰上一角罷了，對RxSwift的學習還須要繼續努力。下面經過一張圖來總結一下：
- 1. RxSwift學習--核心邏輯初探
- 2. RxSwift核心邏輯簡介
- 3. 02.RxSwift - 核心邏輯源碼分析
- 4. RxSwift（2）— 核心邏輯源碼分析
- 5. RxSwift核心邏輯01 -- Observable Observer
- 6. RxSwift （二）序列核心邏輯分析
- 7. RxSwift底層核心邏輯、流程分析
- 8. RxSwift源碼分析(一)-核心邏輯解析
- 9. Spring學習日記十一------MVC核心邏輯篇
- 10. MyBatis 核心邏輯處理篇
- 更多相關文章...
- • ASP.NET Razor - C# 邏輯條件 - ASP.NET 教程
- • ASP.NET Razor - VB 邏輯條件 - ASP.NET 教程
- • Tomcat學習筆記（史上最全tomcat學習筆記）
- • 適用於PHP初學者的學習線路和建議
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. python的安裝和Hello，World編寫
- 2. 重磅解讀：K8s Cluster Autoscaler模塊及對應華爲雲插件Deep Dive
- 3. 鴻蒙學習筆記2（永不斷更）
- 4. static關鍵字 和構造代碼塊
- 5. JVM筆記
- 6. 無法啓動 C/C++ 語言服務器。IntelliSense 功能將被禁用。錯誤: Missing binary at c:\Users\MSI-NB\.vscode\extensions\ms-vsc
- 7. 【Hive】Hive返回碼狀態含義
- 8. Java樹形結構遞歸（以時間換空間）和非遞歸（以空間換時間）
- 9. 數據預處理---缺失值
- 10. 都要2021年了，現代C++有什麼值得我們學習的？