RXSwift中 Subject的使用和分析

Subject 是 RxSwift 中一種特殊的序列, 最重要的特征是 它即為可觀察序列, 也為觀察者. 在實際開發(fā)中也最為常用，因此建議多了解其邏輯原理，對實際開發(fā)使用時有很大幫助。

為什么它是雙重身份呢?
廢話不多說,上代碼

/// Represents an object that is both an observable sequence as well as an observer.
///
/// Each notification is broadcasted to all subscribed observers.
public final class PublishSubject<Element>
    : Observable<Element>
    , SubjectType
    , Cancelable
    , ObserverType
    , SynchronizedUnsubscribeType

它繼承于 Observable 和 ObserverType , 因此 他既有 subscribe 功能,也有 on(_ event: Event<E>) 功能

PublishSubject

  • 使用案例
// 1:初始化序列
let publishSub = PublishSubject<Int>() //初始化一個PublishSubject 裝著Int類型的序列
// 2:發(fā)送響應(yīng)序列
publishSub.onNext(1)
// 3:訂閱序列
publishSub.subscribe { print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發(fā)送響應(yīng)
publishSub.onNext(2)
publishSub.onNext(3)
  • 打印結(jié)果 :

訂閱到了: next(2)
訂閱到了: next(3)

  • 分析:
    cmd + 點擊 PublishSubject 進入該類, 找到 on 方法,中間繼承鏈查找方法流程不在多贅述
public func on(_ event: Event<Element>) {
    #if DEBUG
        self._synchronizationTracker.register(synchronizationErrorMessage: .default)
        defer { self._synchronizationTracker.unregister() }
    #endif
    dispatch(self._synchronized_on(event), event)
}

先來看 self._synchronized_on(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()
    }
}

返回的是當(dāng)前 PublishSubject 的 Observers ,一個 Bag對象,可以理解為數(shù)組, 也就是當(dāng)前觀察者數(shù)組.

再點擊 dispatch 找到如下方法:

func dispatch<E>(_ bag: Bag<(Event<E>) -> Void>, _ event: Event<E>) {
    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)
        }
    }
}

可以看到,該方法是循環(huán)自己保存的所有觀察者,調(diào)用發(fā)送 event ,在第一次執(zhí)行 publishSub.onNext(1) 時, 并沒有觀察者, 因此并無響應(yīng).

則回到代碼中 下一步

publishSub.subscribe { print("訂閱到了:",$0)}

進入 PublishSubject 類尋找 subscribe 方法
過度方法:

public override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
    self._lock.lock()
    let subscription = self._synchronized_subscribe(observer)
    self._lock.unlock()
    return subscription
}

再進入 _synchronized_subscribe

func _synchronized_subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
    if let stoppedEvent = self._stoppedEvent {
        observer.on(stoppedEvent)
        return Disposables.create()
    }
    
    if self._isDisposed {
        observer.on(.error(RxError.disposed(object: self)))
        return Disposables.create()
    }
    
    let key = self._observers.insert(observer.on)
    return SubscriptionDisposable(owner: self, key: key)
}

可以明顯看到 訂閱一次,就在觀察者數(shù)組中插入這個觀察者.

因此, PublishSubject 在未訂閱時,所發(fā)送的 event是沒有響應(yīng)的, 后續(xù)發(fā)送 會查找所有觀察者, 循環(huán)遍歷, 一一發(fā)送響應(yīng).

BehaviorSubject

  • 使用案例
let behaviorSub = BehaviorSubject.init(value: 100)
// 2:發(fā)送信號
behaviorSub.onNext(2)
behaviorSub.onNext(3)
// 3:訂閱序列
behaviorSub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
// 再次發(fā)送
behaviorSub.onNext(4)
behaviorSub.onNext(5)
// 再次訂閱
behaviorSub.subscribe{ print("訂閱到了:",$0)}
    .disposed(by: disposbag)
  • 打印結(jié)果 :

訂閱到了: next(3)
訂閱到了: next(4)
訂閱到了: next(5)
訂閱到了: next(5)

注釋掉 behaviorSub.onNext(2) 和 behaviorSub.onNext(3)
打印結(jié)果:

訂閱到了: next(100)
訂閱到了: next(4)
訂閱到了: next(5)
訂閱到了: next(5)

  • 分析 :
public init(value: Element) {
    self._element = value

    #if TRACE_RESOURCES
        _ = Resources.incrementTotal()
    #endif
}

初始化時,保存了傳進來的默認(rèn)值到自己的 _element 屬性中.
后續(xù)每次發(fā)送 behaviorSub.onNext(2)，來到如下方法

public func on(_ event: Event<E>) {
    dispatch(self._synchronized_on(event), event)
}

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
}

可以看到，跟 PublishSubject 基本差不多，除了給所有觀察者發(fā)送事件以外，多了一步

self._element = element

也就是保存了最新發(fā)送的 element

同樣，subscribe 方法：

public override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
    self._lock.lock()
    let subscription = self._synchronized_subscribe(observer)
    self._lock.unlock()
    return subscription
}

func _synchronized_subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
    
    let key = self._observers.insert(observer.on)
    observer.on(.next(self._element))

    return SubscriptionDisposable(owner: self, key: key)
}

subscribe 方法也是跟 PublishSubject 基本一致，除了把此次觀察者添加保存，多了一步

observer.on(.next(self._element))

也就是說在訂閱時 會默認(rèn)發(fā)送一次 on next 事件，發(fā)送內(nèi)容為保存的最新 element

ReplaySubject

  • 使用案例
    // 1:創(chuàng)建序列
    let replaySub = ReplaySubject<Int>.create(bufferSize: 2)
    // let replaySub = ReplaySubject<Int>.createUnbounded()
    
    // 2:發(fā)送信號
    replaySub.onNext(1)
    replaySub.onNext(2)
    replaySub.onNext(3)
    replaySub.onNext(4)
    
    // 3:訂閱序列
    let disposbag = DisposeBag()
    replaySub.subscribe{ print("訂閱到了:",$0)}
        .disposed(by: disposbag)
    // 再次發(fā)送
    replaySub.onNext(7)
    replaySub.onNext(8)
    replaySub.onNext(9)
  • 打印結(jié)果 :

訂閱到了: next(3)
訂閱到了: next(4)
訂閱到了: next(7)
訂閱到了: next(8)
訂閱到了: next(9)

  • 分析
    ReplaySubject 和 BehaviorSubject 其實實現(xiàn)原理基本類似，只不過后者是用一個 _element 屬性來記錄最新一次的 event 值，而前者在給定的 bufferSize 大于1時使用 _queue 來存儲一組 Element，等于1時 使用一個 _value 屬性來記錄最新一次的 event 值。

AsyncSubject

  • 使用案例
    // AsyncSubject
    // 1:創(chuàng)建序列
    let asynSub = AsyncSubject<Int>.init()
    // 2:發(fā)送信號
    asynSub.onNext(1)
    asynSub.onNext(2)
    // 3:訂閱序列
    let disposbag = DisposeBag()
    asynSub.subscribe{ print("訂閱到了:",$0)}
        .disposed(by: disposbag)
    // 再次發(fā)送
    asynSub.onNext(3)
    asynSub.onNext(4)
//        asynSub.onError(NSError.init(domain: "lb", code: 10086, userInfo: nil))
    asynSub.onCompleted()
  • 打印結(jié)果

訂閱到了: next(4)
訂閱到了: completed

打開注釋掉的 onError這一行

  • 打印結(jié)果

訂閱到了: error(Error Domain=lb Code=10086 "(null)")

  • 分析
    直接查看源碼
public func on(_ event: Event<E>) {
    #if DEBUG
        self._synchronizationTracker.register(synchronizationErrorMessage: .default)
        defer { self._synchronizationTracker.unregister() }
    #endif
    let (observers, event) = self._synchronized_on(event)
    switch event {
    case .next:
        dispatch(observers, event)
        dispatch(observers, .completed)
    case .completed:
        dispatch(observers, event)
    case .error:
        dispatch(observers, event)
    }
}

繼續(xù)點擊 _synchronized_on 方法

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

上面可以看到其實 AsyncSubject 在接收到 on(_ event: Event<E>) 事件時 只有在接收到 .completed 事件時，才會自動發(fā)送一次 .next(lastElement) ，AsyncSubject 在實際開發(fā)中使用的并不多。

AsyncSubject 在完成時發(fā)出最后一個元素，如果源 Observable 沒有發(fā)出任何元素，只有一個完成事件，則 AsyncSubject 也只有一個完成事件。如果源 Observable 產(chǎn)生了一個 error 事件而中止，那么 AsyncSubject 就不會發(fā)出任何元素，而是將 error 事件發(fā)送出來。

Variable

Variable 在5.0版本中已經(jīng)廢棄，使用( BehaviorSubject 或者 BehaviorRelay 替換)

  • 使用案例
    let variableSub = Variable.init(1)
    // 2:發(fā)送信號
    variableSub.value = 100
    variableSub.value = 10
    // 3:訂閱信號
    let disposbag = DisposeBag()
    variableSub.asObservable().subscribe{ print("訂閱到了:",$0)}
        .disposed(by: disposbag)
    // 再次發(fā)送
    variableSub.value = 1000
  • 打印結(jié)果：

訂閱到了: next(10)
訂閱到了: next(1000)

  • 分析

從源碼可以看出，Variable 雖然沒有繼承自 ObserverType 或者 Observable 。但是其有一個 _subject: BehaviorSubject<Element> 屬性。所以，Variable 的行為和 BehaviorSubject 是一致的。但因為不是繼承自 ObserverType，所以沒有 on 函數(shù)，不能直接調(diào)用 on 函數(shù)發(fā)送信號。

在初始化時，使用初始化值，初始化 BehaviorSubject，并保存在 self._subject 中。
對 value 做了一層封裝，在 value 的 set 函數(shù)中，會調(diào)用 _subject 的 on 函數(shù)。完成信號的發(fā)送。

public typealias E = Element

    private let _subject: BehaviorSubject<Element>

    // state
    private var _value: E

    public var value: E {
        get {
            self._lock.lock(); defer { self._lock.unlock() }
            return self._value
        }
        set(newValue) {
            #if DEBUG
                self._synchronizationTracker.register(synchronizationErrorMessage: .variable)
                defer { self._synchronizationTracker.unregister() }
            #endif
            self._lock.lock()
            self._value = newValue
            self._lock.unlock()

            self._subject.on(.next(newValue))
        }
    }

BehaviorRelay

BehaviorRelay 就是 BehaviorSubject 去掉終止事件 onError 和 onCompleted 。

  • 使用案例：
let subject = BehaviorRelay(value: 1)
subject.accept(10)

subject.subscribe({ print("訂閱到：\($0)")})
    .disposed(by: disposeBag)

subject.accept(100)
subject.accept(1000)
  • 打印結(jié)果：

訂閱到：next(10)
訂閱到：next(100)
訂閱到：next(1000)

  • 分析

    查看源碼，請注意 BehaviorRelay 上方的注釋，注釋中說得非常清楚，BehaviorRelay 是對 BehaviorSubject 的封裝，但是和 BehaviorSubject 不一樣的地方在于，BehaviorRelay 不會被 error 和 completed 事件終止。

BehaviorRelay 和 Variable 類似， 無需手寫 on(event)事件，只需給屬性賦值即可拿到響應(yīng)事件內(nèi)容，是我們開發(fā)中最常用的序列。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

  • 發(fā)現(xiàn) 關(guān)注 消息 RxSwift入坑解讀-你所需要知道的各種概念 沸沸騰關(guān)注 2016.11.27 19:11*字...
    楓葉1234閱讀 2,934評論 0贊 2
  • 本文章內(nèi)部分圖片資源來自RayWenderlich.com 本文結(jié)合自己的理解來總結(jié)介紹一下RxSwift最基本的...
    FKSky閱讀 3,029評論 4贊 14
  • 首先,何為RxSwift? RxSwift是ReactiveX的Swift版本,一個響應(yīng)式變成框架。傳送門 開始之...
    cocoawork丶閱讀 550評論 0贊 3
  • 最近比較忙，更新得有點慢，望諒解。 什么是Subject？ 上一章我介紹了Observable——一個功能就像一條...
    turtleeeee閱讀 1,971評論 2贊 14
  • 時至中午，雨露還不曾隱匿陽光而褪去――折射出七色的美麗；螞蟻也去窺視花蕾的裙底，不知是期待它的綻放還是嗅之馨香，久...
    掃云閱讀 510評論 0贊 0

友情鏈接更多精彩內(nèi)容