RxSwift中的Subject

什么是Subject

Rxswift 简单的分能够分为：

• 可观察序列（Observable）
• 观察者（Observer）
• 调度者（Scheduler）
• 销毁者（Dispose）

Subject 既能够作序列，也能够作观察者，因为这个缘由使得 Subject 使用特别方便，在项目中会大量使用。

Rxswift 中主要有4种 Subject：

• PublishSubject
• BehaviorSubject
• ReplaySubject
• AsyncSubject

SubjectType 协议

Rxswift 中全部的 Subject 都会遵循 SubjectType 协议。

• SubjectType 它继承于 ObservableType，它能够做为一个可观察的序列。
• SubjectType 关联了（associatedtype）一个观察者协议（ObserverType）以及方法 asObserver，它能够作为一个观察者来使用。

/// Represents an object that is both an observable sequence as well as an observer.
public protocol SubjectType : ObservableType {
    /// The type of the observer that represents this subject.
    ///
    /// Usually this type is type of subject itself, but it doesn't have to be.
    associatedtype SubjectObserverType : ObserverType

    /// Returns observer interface for subject.
    ///
    /// - returns: Observer interface for subject.
    func asObserver() -> SubjectObserverType
    
}
复制代码

PublishSubject

/// 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 {
    
   ...(省略)
}
复制代码

1. 继承于 Observable<Element>（可被观察序列），遵循协议：
   • SubjectType —— 既能够做为 Observerable 又可做为 Observer
   • Cancelable —— 能够被销毁（继承于 Disposable）
   • ObserverType —— 观察者
   • SynchronizedUnsubscribeType —— 取消订阅
2. PublishSubject 只能接收在被订阅以后发送的信号

示例：

let pub = PublishSubject<Int>()

pub.onNext(1)

pub.subscribe(onNext: { (num) in
    print("接收到——\(num)")
}).disposed(by: disposeBag)

pub.onNext(2)
pub.onNext(3)
复制代码

输出：

接收到——2
接收到——3
复制代码

BehaviorSubject

/// Represents a value that changes over time.
///
/// Observers can subscribe to the subject to receive the last (or initial) value and all subsequent notifications.
public final class BehaviorSubject<Element> : Observable<Element> , SubjectType , ObserverType , SynchronizedUnsubscribeType , Cancelable {
    
    ...(省略)
}
复制代码

1. 继承于 Observable<Element>（可被观察序列），遵循协议：

   • SubjectType —— 既能够做为 Observerable 又可做为 Observer
   • Cancelable —— 能够被销毁（继承于 Disposable）
   • ObserverType —— 观察者
   • SynchronizedUnsubscribeType —— 取消订阅

2. 经过一个初始默认值建立，会存储上一次信号

// 初始化默认传入一个值
    public init(value: Element) {
        self._element = value

        #if TRACE_RESOURCES
            _ = Resources.incrementTotal()
        #endif
    }
复制代码

func _synchronized_on(_ event: Event<E>) -> Observers {
        self._lock.lock(); defer { self._lock.unlock() }
        if self._stoppedEvent != nil || self._isDisposed {
            return Observers()
        }
        
        switch event {
        case .next(let element):
            // 记录值
            self._element = element
        case .error, .completed:
            self._stoppedEvent = event
        }
        
        return self._observers
    }
复制代码

三、订阅以后首先会接收到最近一次发送的事件（若是最近没有发送，那么发送一个初始的事件）

示例：

let behavior = BehaviorSubject<Int>(value: 1)

behavior.onNext(2)

behavior.subscribe(onNext: { (num) in
    print("接收到——\(num)")
}).disposed(by: disposeBag)

behavior.onNext(3)
behavior.onNext(4)

let value = try? behavior.value()
print("value = \(value)")
复制代码

输出：

接收到——2
接收到——3
接收到——4
value = Optional(4)
复制代码

ReplaySubject

/// Represents an object that is both an observable sequence as well as an observer.
///
/// Each notification is broadcasted to all subscribed and future observers, subject to buffer trimming policies.
public class ReplaySubject<Element> : Observable<Element> , SubjectType , ObserverType , Disposable {
    
    ...(省略)
}
复制代码

1. 继承于 Observable<Element>（可被观察序列），遵循协议：
   • SubjectType —— 既能够做为 Observerable 又可做为 Observer
   • ObserverType —— 观察者
   • Disposable —— 能够被销毁
2. ReplaySubject 会根据缓冲区调整，每一个信号将广播给全部已订阅的观察者和将来订阅的观察者。

示例：

let replay = ReplaySubject<Int>.create(bufferSize: 2)

replay.onNext(1)
replay.onNext(2)
replay.onNext(3)

replay.subscribe(onNext: { (num) in
    print("第一个订阅接收到 -- \(num)")
}).disposed(by: disposeBag)

replay.onNext(4)
replay.onNext(5)

replay.subscribe(onNext: { (num) in
    print("第二个订阅接收到 -- \(num)")
}).disposed(by: disposeBag)
复制代码

输出：

第一个订阅接收到 -- 2
第一个订阅接收到 -- 3
第一个订阅接收到 -- 4
第一个订阅接收到 -- 5
第二个订阅接收到 -- 4
第二个订阅接收到 -- 5
复制代码

AsyncSubject

public final class AsyncSubject<Element> : Observable<Element> , SubjectType , ObserverType , SynchronizedUnsubscribeType {
    
    ...（省略）
}
复制代码

1. 继承于 Observable<Element>（可被观察序列），遵循协议：

   • SubjectType —— 既能够做为 Observerable 又可做为 Observer
   • ObserverType —— 观察者
   • SynchronizedUnsubscribeType —— 取消订阅

2. AsyncSubject 只发送最后一个信号，而且只在完成（onCompleted）以后。

示例1：

let async = AsyncSubject<Int>()

async.onNext(1)
async.onNext(2)

async.subscribe(onNext: { (num) in
    print("接受订阅 = \(num)")
}).disposed(by: disposeBag)

async.onNext(3)
async.onNext(4)

async.onCompleted()
复制代码

输出：

接受订阅 = 4
复制代码

RxCocoa 中的 PublishRelay, BehaviorRelay

• PublishRelay 是对 PublishSubject 的封装
• BehaviorRelay 是对 BehaviorSubject 的封装
• PublishSubject 和 BehaviorSubject 都会被 error 或者 completed 便可以用方法 onError 或者 onCompleted 终止；可是 PublishRelay 和 BehaviorRelay 不会（就没有 onError 或者 onCompleted方法）

/// PublishRelay is a wrapper for `PublishSubject`.
///
/// Unlike `PublishSubject` it can't terminate with error or completed.
public final class PublishRelay<Element>: ObservableType {
    public typealias E = Element

    private let _subject: PublishSubject<Element>
    
    ...（省略）
}
复制代码

/// BehaviorRelay is a wrapper for `BehaviorSubject`.
///
/// Unlike `BehaviorSubject` it can't terminate with error or completed.
public final class BehaviorRelay<Element>: ObservableType {
    public typealias E = Element

    private let _subject: BehaviorSubject<Element>
    
    ...（省略）
}
复制代码

PublishRelay 示例：

let publisRelay = PublishRelay<Int>()

publisRelay.accept(1)

publisRelay.subscribe(onNext: { (num) in
    print("接收到 -- \(num)")
}).disposed(by: disposeBag)

publisRelay.accept(2)
publisRelay.accept(3)
复制代码

输出：

接收到 -- 2
接收到 -- 3
复制代码

BehaviorRelay 示例：

let behaviorRelay = BehaviorRelay<Int>(value: 1)

behaviorRelay.accept(2)

behaviorRelay.subscribe(onNext: { (num) in
    print("接收到 -- \(num)")
}).disposed(by: disposeBag)

behaviorRelay.accept(3)
behaviorRelay.accept(4)

print("value = \(behaviorRelay.value)")
复制代码

输出：

接收到 -- 2
接收到 -- 3
接收到 -- 4
value = 4
复制代码