14.RxSwift 訂閱者(Subject)

publish()函數的本質

在高階函數四中最後一節multicast的時候寫過一個例子：原序列被publish修飾過

這裏 publish()有什麼騷操做呢：返回了一個 PublishSubject

multicast裏調用 ConnectableObservableAdapter 返回了一個 lazySubject:

而且這個 lazySubject執行了訂閱：

這裏的 subscribe並無執行 sink,也就不會建立觀察者 observer,沒有觀察者，不是就不能發送 on函數，最終也就不能完成回調，那麼這裏是如何關聯建立了 observer的呢，請看

這裏的 connect()作了啥

這裏只添加了一個觀察者(中間臨時變量) connection，調用自身 on必然就會發起 _subjectObserver.on也就意味着發起了響應

而這裏又是臨時變量，顧只會調用一次，也就是隻會發起一次響應，最終咱們也就只會響應一次訂閱

Subject

在RxsWift中還有一種很是特殊的序列Subject- 即公也爲受 Subject是一個代理，它既是``Observer，也是 Observable 以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)
複製代碼

這裏只打印2和3，緣由是它只接收被訂閱subscribe以後發送的onNext

1.建立Subject類 2.收集觀察者回調 3.發送信號(遍歷集合observers調用on)

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

複製代碼

這裏只打印3,4,5，和PublishSubject相比會保留訂閱前的最後一次發送，即：

• ReplaySubject

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)

複製代碼

會打印3，4，7，8，9和BehaviorSubject相比又多保留一次，即保留兩次，這裏保存集合被指定大小爲bufferSize=2

• 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()
複製代碼

• Variable

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

複製代碼

BehaviorSubject替換使用: