Subject 是 RxSwift 中一种特殊的序列, 最重要的特征是 它即为可观察序列, 也为观察者. 在实际开发中也最为常用，因此建议多了解其逻辑原理，对实际开发使用时有很大帮助。
为什么它是双重身份呢?
废话不多说,上代码
/// 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
它继承于 Observable 和 ObserverType , 因此 他既有 subscribe 功能,也有 on(_ event: Event<E>) 功能
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)
- 打印结果 :
订阅到了: next(2)
订阅到了: next(3)
- 分析:
cmd + 点击PublishSubject进入该类, 找到on方法,中间继承链查找方法流程不在多赘述
public func on(_ event: Event<Element>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
dispatch(self._synchronized_on(event), event)
}
先来看 self._synchronized_on(event)
func _synchronized_on(_ event: Event<E>) -> Observers {
self._lock.lock(); defer { self._lock.unlock() }
switch event {
case .next:
if self._isDisposed || self._stopped {
return Observers()
}
return self._observers
case .completed, .error:
if self._stoppedEvent == nil {
self._stoppedEvent = event
self._stopped = true
let observers = self._observers
self._observers.removeAll()
return observers
}
return Observers()
}
}
返回的是当前 PublishSubject 的 Observers ,一个 Bag对象,可以理解为数组, 也就是当前观察者数组.
再点击 dispatch 找到如下方法:
func dispatch<E>(_ bag: Bag<(Event<E>) -> Void>, _ event: Event<E>) {
bag._value0?(event)
if bag._onlyFastPath {
return
}
let pairs = bag._pairs
for i in 0 ..< pairs.count {
pairs[i].value(event)
}
if let dictionary = bag._dictionary {
for element in dictionary.values {
element(event)
}
}
}
可以看到,该方法是循环自己保存的所有观察者,调用发送 event ,在第一次执行 publishSub.onNext(1) 时, 并没有观察者, 因此并无响应.
则回到代码中 下一步
publishSub.subscribe { print("订阅到了:",$0)}
进入 PublishSubject 类寻找 subscribe 方法
过度方法:
public override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
self._lock.lock()
let subscription = self._synchronized_subscribe(observer)
self._lock.unlock()
return subscription
}
再进入 _synchronized_subscribe
func _synchronized_subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
if let stoppedEvent = self._stoppedEvent {
observer.on(stoppedEvent)
return Disposables.create()
}
if self._isDisposed {
observer.on(.error(RxError.disposed(object: self)))
return Disposables.create()
}
let key = self._observers.insert(observer.on)
return SubscriptionDisposable(owner: self, key: key)
}
可以明显看到 订阅一次,就在观察者数组中插入这个观察者.
因此,
PublishSubject在未订阅时,所发送的 event是没有响应的, 后续发送 会查找所有观察者, 循环遍历, 一一发送响应.
BehaviorSubject
- 使用案例
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)
- 打印结果 :
订阅到了: next(3)
订阅到了: next(4)
订阅到了: next(5)
订阅到了: next(5)
注释掉 behaviorSub.onNext(2) 和 behaviorSub.onNext(3)
打印结果:
订阅到了: next(100)
订阅到了: next(4)
订阅到了: next(5)
订阅到了: next(5)
- 分析 :
public init(value: Element) {
self._element = value
#if TRACE_RESOURCES
_ = Resources.incrementTotal()
#endif
}
初始化时,保存了传进来的默认值到自己的 _element 属性中.
后续每次发送 behaviorSub.onNext(2)，来到如下方法
public func on(_ event: Event<E>) {
dispatch(self._synchronized_on(event), event)
}
func _synchronized_on(_ event: Event<E>) -> Observers {
switch event {
case .next(let element):
self._element = element
case .error, .completed:
self._stoppedEvent = event
}
return self._observers
}
可以看到，跟 PublishSubject 基本差不多，除了给所有观察者发送事件以外，多了一步
self._element = element
也就是保存了最新发送的 element
同样，subscribe 方法：
public override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
self._lock.lock()
let subscription = self._synchronized_subscribe(observer)
self._lock.unlock()
return subscription
}
func _synchronized_subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
let key = self._observers.insert(observer.on)
observer.on(.next(self._element))
return SubscriptionDisposable(owner: self, key: key)
}
subscribe 方法也是跟 PublishSubject 基本一致，除了把此次观察者添加保存，多了一步
observer.on(.next(self._element))
也就是说在订阅时 会默认发送一次 on next 事件，发送内容为保存的最新 element
ReplaySubject
- 使用案例
// 1:创建序列
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:订阅序列
let disposbag = DisposeBag()
replaySub.subscribe{ print("订阅到了:",$0)}
.disposed(by: disposbag)
// 再次发送
replaySub.onNext(7)
replaySub.onNext(8)
replaySub.onNext(9)
- 打印结果 :
订阅到了: next(3)
订阅到了: next(4)
订阅到了: next(7)
订阅到了: next(8)
订阅到了: next(9)
- 分析
ReplaySubject和BehaviorSubject其实实现原理基本类似，只不过后者是用一个_element属性来记录最新一次的 event 值，而前者在给定的bufferSize大于1时使用_queue来存储一组Element，等于1时 使用一个_value属性来记录最新一次的 event 值。
AsyncSubject
- 使用案例
// AsyncSubject
// 1:创建序列
let asynSub = AsyncSubject<Int>.init()
// 2:发送信号
asynSub.onNext(1)
asynSub.onNext(2)
// 3:订阅序列
let disposbag = DisposeBag()
asynSub.subscribe{ print("订阅到了:",$0)}
.disposed(by: disposbag)
// 再次发送
asynSub.onNext(3)
asynSub.onNext(4)
// asynSub.onError(NSError.init(domain: "lb", code: 10086, userInfo: nil))
asynSub.onCompleted()
- 打印结果
订阅到了: next(4)
订阅到了: completed
打开注释掉的 onError这一行
- 打印结果
订阅到了: error(Error Domain=lb Code=10086 "(null)")
- 分析
直接查看源码
public func on(_ event: Event<E>) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .default)
defer { self._synchronizationTracker.unregister() }
#endif
let (observers, event) = self._synchronized_on(event)
switch event {
case .next:
dispatch(observers, event)
dispatch(observers, .completed)
case .completed:
dispatch(observers, event)
case .error:
dispatch(observers, event)
}
}
继续点击 _synchronized_on 方法
func _synchronized_on(_ event: Event<E>) -> (Observers, Event<E>) {
switch event {
case .next(let element):
self._lastElement = element
return (Observers(), .completed)
case .error:
self._stoppedEvent = event
let observers = self._observers
self._observers.removeAll()
return (observers, event)
case .completed:
let observers = self._observers
self._observers.removeAll()
if let lastElement = self._lastElement {
self._stoppedEvent = .next(lastElement)
return (observers, .next(lastElement))
}
else {
self._stoppedEvent = event
return (observers, .completed)
}
}
}
上面可以看到其实 AsyncSubject 在接收到 on(_ event: Event<E>) 事件时 只有在接收到 .completed 事件时，才会自动发送一次 .next(lastElement) ，AsyncSubject 在实际开发中使用的并不多。
AsyncSubject 在完成时发出最后一个元素，如果源 Observable 没有发出任何元素，只有一个完成事件，则 AsyncSubject 也只有一个完成事件。如果源 Observable 产生了一个 error 事件而中止，那么 AsyncSubject 就不会发出任何元素，而是将 error 事件发送出来。
Variable
Variable 在5.0版本中已经废弃，使用( BehaviorSubject 或者 BehaviorRelay 替换)
- 使用案例
let variableSub = Variable.init(1)
// 2:发送信号
variableSub.value = 100
variableSub.value = 10
// 3:订阅信号
let disposbag = DisposeBag()
variableSub.asObservable().subscribe{ print("订阅到了:",$0)}
.disposed(by: disposbag)
// 再次发送
variableSub.value = 1000
- 打印结果：
订阅到了: next(10)
订阅到了: next(1000)
- 分析
从源码可以看出，Variable 虽然没有继承自 ObserverType 或者 Observable 。但是其有一个 _subject: BehaviorSubject<Element> 属性。所以，Variable 的行为和 BehaviorSubject 是一致的。但因为不是继承自 ObserverType，所以没有 on 函数，不能直接调用 on 函数发送信号。
在初始化时，使用初始化值，初始化 BehaviorSubject，并保存在 self._subject 中。
对 value 做了一层封装，在 value 的 set 函数中，会调用 _subject 的 on 函数。完成信号的发送。
public typealias E = Element
private let _subject: BehaviorSubject<Element>
// state
private var _value: E
public var value: E {
get {
self._lock.lock(); defer { self._lock.unlock() }
return self._value
}
set(newValue) {
#if DEBUG
self._synchronizationTracker.register(synchronizationErrorMessage: .variable)
defer { self._synchronizationTracker.unregister() }
#endif
self._lock.lock()
self._value = newValue
self._lock.unlock()
self._subject.on(.next(newValue))
}
}
BehaviorRelay
BehaviorRelay就是BehaviorSubject去掉终止事件onError和onCompleted。
- 使用案例：
let subject = BehaviorRelay(value: 1)
subject.accept(10)
subject.subscribe({ print("订阅到：\($0)")})
.disposed(by: disposeBag)
subject.accept(100)
subject.accept(1000)
- 打印结果：
订阅到：next(10)
订阅到：next(100)
订阅到：next(1000)
- 分析
查看源码，请注意BehaviorRelay上方的注释，注释中说得非常清楚，BehaviorRelay是对BehaviorSubject的封装，但是和BehaviorSubject不一样的地方在于，BehaviorRelay不会被error和completed事件终止。
BehaviorRelay和Variable类似， 无需手写 on(event)事件，只需给属性赋值即可拿到响应事件内容，是我们开发中最常用的序列。
