14.RxSwift 订阅者(Subject)

156 阅读1分钟
publish()函数的本质

在高阶函数四中最后一节multicast的时候写过一个例子:原序列被publish修饰过

这里publish()有什么骚操作呢:返回了一个PublishSubject
multicast里调用ConnectableObservableAdapter 返回了一个lazySubject:

并且这个lazySubject执行了订阅:

这里的subscribe并没有执行sink,也就不会创建观察者observer,没有观察者,不是就不能发送on函数,最终也就不能完成回调,那么这里是如何关联创建了observer的呢,请看

这里的connect()做了啥

这里只添加了一个观察者(中间临时变量)connection,调用自身on必然就会发起_subjectObserver.on也就意味着发起了响应

而这里又是临时变量,顾只会调用一次,也就是只会发起一次响应,最终我们也就只会响应一次订阅

Subject

RxsWift中还有一种非常特殊的序列Subject- 即公也为受 Subject是一个代理,它既是``Observer,也是 ObservablePublishSubject为例:

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