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替换使用: