RxSwift（12）— Subject即攻也守

就問此時此刻還有誰？45度仰望天空，該死！我這無處安放的魅力！

• RxSwift（1）— 初探
• RxSwift（2）— 核心邏輯源碼分析
• RxSwift（3）— Observable序列的建立方式
• RxSwift（4）— 高階函數（上）
• RxSwift（5）— 高階函數（下）
• RxSwift（6）— 調度者-scheduler源碼解析（上）
• RxSwift（7）— 調度者-scheduler源碼解析（下）
• RxSwift（8）— KVO底層探索（上）
• RxSwift（9）— KVO底層探索（下）
• RxSwift（10）— 場景序列總結
• RxSwift（11）— dispose源碼解析
• RxSwift（12）— Subject即攻也守
• RxSwift（13）— 爬過的坑
• RxSwift（14）— MVVM雙向綁定

---

RxSwift 目錄直通車 --- 和諧學習，不急不躁！

---

在掌握前面序列以還有觀察者的前提下，咱們今天來看一個很是特殊的類型-Subject.爲何說它特殊呢？緣由很簡單：**Subject既能夠作序列，也能夠作觀察者！**正是由於這一特性，因此在實際開發中被大量運用。下面咱們一塊兒來解讀一下這個特殊的Subject

即攻也守的原理

首先咱們來看看：SubjectType的原理！

public protocol SubjectType : ObservableType {
      // 關聯了觀察者類型，具有這個類型的能力
    associatedtype SubjectObserverType : ObserverType
    func asObserver() -> SubjectObserverType
}
複製代碼

• SubjectType首先就是繼承了ObservableType，具備序列特性
• 關聯了觀察者類型，具有這個類型的能力
• 下面咱們經過一個具體類型來感覺一下subject

// 1:初始化序列
let publishSub = PublishSubject<Int>() 
// 2:發送響應序列
publishSub.onNext(1)
// 3:訂閱序列
publishSub.subscribe { print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送響應
publishSub.onNext(2)
publishSub.onNext(3)
複製代碼

• 很明顯可以訂閱信號（序列最基本的能力）
• 可以發送響應，又是觀察者的能力
• 查看底層源碼分析

訂閱響應流程

public override func subscribe -> Disposable {
    self._lock.lock()
    let subscription = self._synchronized_subscribe(observer)
    self._lock.unlock()
    return subscription
}

func _synchronized_subscribe -> Disposable {
    // 省略沒必要要的代碼
    let key = self._observers.insert(observer.on)
    return SubscriptionDisposable(owner: self, key: key)
}
複製代碼

• self._observers.insert(observer.on): 經過一個集合添加進去全部的訂閱事件，很明顯在合適的地方一次性所有執行
• 其中也返回此次訂閱的銷燬者，方便執行善後工做： synchronizedUnsubscribe->self._observers.removeKey(disposeKey)

mutating func removeKey(_ key: BagKey) -> T? {
    if _key0 == key {
        _key0 = nil
        let value = _value0!
        _value0 = nil
        return value
    }

    if let existingObject = _dictionary?.removeValue(forKey: key) {
        return existingObject
    }

    for i in 0 ..< _pairs.count where _pairs[i].key == key {
        let value = _pairs[i].value
        _pairs.remove(at: i)
        return value
    }
    return nil
}
複製代碼

• 便利經過key獲取響應bag中的value,執行集合移除
• 由於沒有相應持有關係，達到自動釋放銷燬

發送信號流程

public func on(_ event: Event<Element>) {
        dispatch(self._synchronized_on(event), event)
    }
複製代碼

• 這個地方估計你們看起來麻煩噁心一點，可是你用心看不難體會
• 這裏主要調用了dispatch函數，傳了兩個參數：self._synchronized_on(event)和event
• 查看dispatch函數源碼

func dispatch<E>(_ bag: Bag) {
    bag._value0?(event)

    if bag._onlyFastPath {
        return
    }

    let pairs = bag._pairs
    for i in 0 ..< pairs.count {
        pairs[i].value(event)
    }

    if let dictionary = bag._dictionary {
        for element in dictionary.values {
            element(event)
        }
    }
}
複製代碼

• bag._value0?(event)首先執行事件的回調
• 判斷bag._onlyFastPath的狀況，默認會開啓快速通道！
• 若是是開啓慢速通道，須要從剛剛添加進bag包裹裏面的匹配對挨個進行pairs[i].value(event),外界事件回調，而後拿回外界封裝的閉包的閉包調用:element(event)

func _synchronized_on(_ event: Event<E>) -> Observers {
    self._lock.lock(); defer { self._lock.unlock() }
    switch event {
    case .next:
        if self._isDisposed || self._stopped {
            return Observers()
        }
        
        return self._observers
    case .completed, .error:
        if self._stoppedEvent == nil {
            self._stoppedEvent = event
            self._stopped = true
            let observers = self._observers
            self._observers.removeAll()
            return observers
        }

        return Observers()
    }
}
複製代碼

• 這裏若是self._isDisposed || self._stopped成立就會返回一個空的集合，也就沒有序列的響應
• 在.completed, .error都會改變狀態self._stopped = true,也就是說序列完成或者錯誤以後都沒法再次響應了
• 在.completed, .error還會移除添加在集合裏面的內容

其實若是你對前面序列的流程掌握了，這個subject的流程也再也不話下,只是subject 把訂閱流程和響應流程都內部實現，因此也就沒有必要引入sink

各類Subject

PublishSubject

能夠不須要初始來進行初始化（也就是能夠爲空），而且它只會向訂閱者發送在訂閱以後才接收到的元素。

// PublishSubject
// 1:初始化序列
let publishSub = PublishSubject<Int>() //初始化一個PublishSubject 裝着Int類型的序列
// 2:發送響應序列
publishSub.onNext(1)
// 3:訂閱序列
publishSub.subscribe { print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送響應
publishSub.onNext(2)
publishSub.onNext(3)
複製代碼

• 信號:1是沒法被訂閱的，只接受訂閱以後的響應

BehaviorSubject

經過一個默認初始值來建立，當訂閱者訂閱BehaviorSubject時，會收到訂閱後Subject上一個發出的Event，若是尚未收到任何數據，會發出一個默認值。以後就和PublishSubject同樣，正常接收新的事件。

和publish 稍微不一樣就是behavior這個傢伙有個存儲功能：存儲上一次的信號

// BehaviorSubject
// 1:建立序列
let behaviorSub = BehaviorSubject.init(value: 100)
// 2:發送信號
behaviorSub.onNext(2)
behaviorSub.onNext(3)
// 3:訂閱序列
behaviorSub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送
behaviorSub.onNext(4)
behaviorSub.onNext(5)
// 再次訂閱
behaviorSub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
複製代碼

• 當沒有信號的時候，會默認發送 信號：100
• 只能儲存一個信號：信號2 會被 信號3 覆蓋
• 訂閱信號以前可以儲存信號

// 初始化
public init(value: Element) {
      self._element = value
}

// 事件響應
func _synchronized_on(_ event: Event<E>) -> Observers {

    switch event {
    case .next(let element):
        self._element = element
    case .error, .completed:
        self._stoppedEvent = event
    }
    return self._observers
}
複製代碼

• 初始化的時候帶有一個屬性保存一個信號
• 事件響應：新事件會覆蓋原來的事件
• 其餘流程和publish同樣

ReplaySubject

ReplaySubject 發送源Observable 的全部事件不管observer何時開始訂閱。

// ReplaySubject
// 1:建立序列
let replaySub = ReplaySubject<Int>.create(bufferSize: 2)
// let replaySub = ReplaySubject<Int>.createUnbounded()

// 2:發送信號
replaySub.onNext(1)
replaySub.onNext(2)
replaySub.onNext(3)
replaySub.onNext(4)

// 3:訂閱序列
replaySub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送
replaySub.onNext(7)
replaySub.onNext(8)
replaySub.onNext(9)
複製代碼

• 一個bufferSize空間，想存儲多少次響應就是多少次
• 其餘流程照舊
• 源碼裏面就是相對於BehaviorSubject的儲存屬性變成了集合

AsyncSubject

AsyncSubject只發送由源Observable發送的最後一個事件，而且只在源Observable完成以後。（若是源Observable沒有發送任何值，AsyncSubject也不會發送任何值。）

// AsyncSubject
// 1:建立序列
let asynSub = AsyncSubject<Int>.init()
// 2:發送信號
asynSub.onNext(1)
asynSub.onNext(2)
// 3:訂閱序列
asynSub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送
asynSub.onNext(3)
asynSub.onNext(4)
// asynSub.onError(NSError.init(domain: "lgcooci", code: 10086, userInfo: nil))
asynSub.onCompleted()
複製代碼

• 咱們普通序列發送回來，都不會響應！直到完成序列響應

func _synchronized_on(_ event: Event<E>) -> (Observers, Event<E>) {
    switch event {
    case .next(let element):
        self._lastElement = element
        return (Observers(), .completed)
    case .error:
        self._stoppedEvent = event

        let observers = self._observers
        self._observers.removeAll()

        return (observers, event)
    case .completed:

        let observers = self._observers
        self._observers.removeAll()

        if let lastElement = self._lastElement {
            self._stoppedEvent = .next(lastElement)
            return (observers, .next(lastElement))
        }
        else {
            self._stoppedEvent = event
            return (observers, .completed)
        }
    }
}
複製代碼

• 能夠很清晰的看出，普通Next事件都是，元素的替換，根本沒有響應出來 *complete事件發送到時候，就會把最新保存的self._lastElement當成事件值傳出去，響應.next(lastElement)
• 若是沒有保存事件就發送完成事件：.completed
• error事件會移空整個響應集合：self._observers.removeAll()

Variable

Variable廢棄了，這裏貼出代碼以供你們遇到老版本! 因爲這個Variable的靈活性因此在開發裏面應用很是之多！

// Variable : 5.0已經廢棄(BehaviorSubject 替換) - 這裏板書 你們能夠了解一下
// 1:建立序列
let variableSub = Variable.init(1)
// 2:發送信號
variableSub.value = 100
variableSub.value = 10
// 3:訂閱信號 })

variableSub.asObservable().subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發送
variableSub.value = 1000
複製代碼

BehaviorRelay

• 替換原來的Variable
• 能夠儲存一個信號
• 隨時訂閱響應
• 響應發送的時候要注意：behaviorR.accept(20)

let behaviorRelay = BehaviorRelay(value: 100)
behaviorRelay.subscribe(onNext: { (num) in
    print(num)
.disposed(by: disposbag)
print("打印:\(behaviorRelay.value)")

behaviorRelay.accept(1000)
複製代碼

Subject在實際開發中，應用很是的普遍！平時不少時候都會在惆悵選擇什麼序列更合適，那麼聰明的你必定要掌握底層的原理，並不說你背下特點就能真正開發的，由於若是後面一旦發生了BUG，你根本沒法解決。做爲iOS中高級發開人員**必定要知其然，而知其因此然！**碌碌無爲的應用層開發畢竟走不長遠!

就問此時此刻還有誰？45度仰望天空，該死！我這無處安放的魅力！