RxSwift学习笔记之Subject

本文为原创文章，转载请标明出处

AsyncSubject

一个AsyncSubject只在原始Observable完成后，发射来自原始Observable的最后一个值。它会把这最后一个值发射给任何后续的观察者。

        let disposeBag = DisposeBag()

        let subject = AsyncSubject<Int>()

        subject.onNext(1)

        subject.subscribe(onNext: { int in
            print("observerA: (int)")
        }, onCompleted: {
            print("observerA: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(2)

        subject.subscribe(onNext: { int in
            print("observerB: (int)")
        }, onCompleted: {
            print("observerB: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(3)

        subject.subscribe(onNext: { int in
            print("observerC: (int)")
        }, onCompleted: {
            print("observerC: onCompleted")
        }).disposed(by: disposeBag)

        subject.onCompleted()

        subject.onNext(4)

        subject.subscribe(onNext: { int in
            print("observerD: (int)")
        }, onCompleted: {
            print("observerD: onCompleted")
        }).disposed(by: disposeBag)

输出：

observerA: 3
observerB: 3
observerC: 3
observerA: onCompleted
observerB: onCompleted
observerC: onCompleted
observerD: 3
observerD: onCompleted

BehaviorSubject

当观察者订阅BehaviorSubject时，它开始发射原始Observable最近发射的数据，然后继续发射其它任何来自原始Observable的数据。

        let disposeBag = DisposeBag()

        let subject = BehaviorSubject<Int>(value: 0)

        subject.onNext(1)

        subject.subscribe(onNext: { int in
            print("observerA: (int)")
        }, onCompleted: {
            print("observerA: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(2)

        subject.subscribe(onNext: { int in
            print("observerB: (int)")
        }, onCompleted: {
            print("observerB: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(3)

输出：

observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

PublishSubject

PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。

        let disposeBag = DisposeBag()

        let subject = PublishSubject<Int>()

        subject.onNext(0)

        subject.subscribe(onNext: { int in
            print("observerA: (int)")
        }, onCompleted: {
            print("observerA: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(1)

        subject.subscribe(onNext: { int in
            print("observerB: (int)")
        }, onCompleted: {
            print("observerB: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(2)

输出：

observerA: 1
observerA: 2
observerB: 2

ReplaySubject

ReplaySubject在缓存增长到一定大小后会丢弃旧的数据，不然会发射所有来自原始Observable的数据给观察者，无论它们是何时订阅的。

        let disposeBag = DisposeBag()

        let subject = ReplaySubject<Int>.create(bufferSize: 3)

        subject.onNext(1)
        subject.onNext(2)
        subject.onNext(3)

        subject.subscribe(onNext: { int in
            print("observerA: (int)")
        }, onCompleted: {
            print("observerA: onCompleted")
        }).disposed(by: disposeBag)

        subject.onNext(4)

        subject.subscribe(onNext: { int in
            print("observerB: (int)")
        }, onCompleted: {
            print("observerB: onCompleted")
        }).disposed(by: disposeBag)

        subject.onCompleted()

        subject.subscribe(onNext: { int in
            print("observerC: (int)")
        }, onCompleted: {
            print("observerC: onCompleted")
        }).disposed(by: disposeBag)

输出：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerB: 2
observerB: 3
observerB: 4
observerA: onCompleted
observerB: onCompleted
observerC: 2
observerC: 3
observerC: 4
observerC: onCompleted