前言
RxSwift中的Subjects非常特殊，因为它即充当了观察者序列(Observable)的角色，还充当了观察者(Observer)的角色。
本文的探索比较简单，如果之前有探索过RxSwift核心逻辑的话，那再看本文是相当容易的。
PublishSubject
PublishSubject:一开始的时候是空的，只发送新的元素给观察者。它仅仅会发送observer订阅之后的事件，也就是说如果sequence上有.next的到来，但是这个时候某个observer还没有subscribe它，这个observer就收不到这条信息，它只会收到它订阅之后发生的事件。
简单来说:就是订阅后才接收发出的新的事件，之前的统统不管。
如果不理解，直接上代码
// 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)" ,之前发送的信号1已经不管了，很容易理解的。
探索原理
- 直接点
subscribe进去什么都看不到，当然，作为一个合格的开发者，那就进入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
}
传递了observer到它下面这个方法，关键代码self._observers.insert(observer.on)，向当前的observers集合插入一个observer.on
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)
}
找到它的声明，Observers(）是 AnyObserver<Element>.s的别名,点进去是可以看到的，我们了解到subscribe其实就做了插入一个observer.on保存到_observers里。
private var _observers = Observers()
- 我们再看发送响应的代码，找到
on方法, 调用了这里dispatch(self._synchronized_on(event), event)
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,其实就是在 .next的时候返回当前的所有Observers,.completed或.error的时候，同样返回当前的所有Observers，但是返回之前清空了Observers
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()
}
}
再找到dispatch, 再结合上面返回了当前所有Observers，下面的代码其实就是把Observers里之前存储的observer.on一个一个执行
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)
}
}
}
总结: 探索到了这里，已经完全明白了PulishSubject的原理了，在订阅前发送.next响应，observer都还没有insert,当然接收不到响应咯。所以PublishSubject的好处就是在需要用的时候才去订阅它，不用管之前的发送。
BehaviorSubject
BehaviorSubject和PublishSubject类似，只是它可以保存上一次发送的信号，所以它会重发上一次最近的消息给新的订阅者。
// 1:创建序列
let behaviorSub = BehaviorSubject.init(value: 100)
// 2:发送信号
behaviorSub.onNext(1)
behaviorSub.onNext(2)
// 3:订阅序列
behaviorSub.subscribe{ print("第一次订阅到了:",$0)}
.disposed(by: disposbag)
// 再次发送
behaviorSub.onNext(3)
behaviorSub.onNext(4)
// 再次订阅
behaviorSub.subscribe{ print("第二次订阅到了:",$0)}
.disposed(by: disposbag)
打印结果: 可以看出，第一次订阅接收到了订阅前就发送的事件2,第二次订阅接收到了订阅前就发送的事件4
第一次订阅到了: next(2)
第一次订阅到了: next(3)
第一次订阅到了: next(4)
第二次订阅到了: next(4)
探索原理
由于和PublishSubject类似，轻车熟路找到_synchronized_subscribe,它在插入observer.on到_observers后，还调用了on方法直接执行调用self._element
func _synchronized_subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == E {
//只贴了这关键代码
let key = self._observers.insert(observer.on)
observer.on(.next(self._element))
}
同样再直接看_synchronized_on,在调用.next的时候，还保存了一下本次发送的element，如果之前没有，就是保存，如果之前有，就是更新，反正就是保存最新的一条响应元素
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
}
总结: 和PublishSubject基本相同，只是多创建了一个中间变量Element
来保存最新一次响应的element。一般会使用在一些视图，如果想要获取上一次显示最近的数据是非常有用的。比如加载XX列表，先显示上一次的缓存数据，网络请求完数据再更新数据。
ReplaySubject
其实这个ReplaySubject也和上面的两种类似，只是它创建了一个缓冲区，它可以保存在订阅之前的响应的N个事件，可以自定义缓冲区的大小，想保存几个就几个。
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)
设置了缓冲区为2，所以保存了前2次响应事件, 一订阅就会接收到之前的事件。
打印结果:
订阅到了: next(3)
订阅到了: next(4)
原理总结:其实就是保存了一个queue作为缓冲区，在on的时候，就enqueue把值插入到队列里，在subscrib的时候，就执行一遍queue里保存的事件。
private class ReplayManyBase<Element>: ReplayBufferBase<Element> {
fileprivate var _queue: Queue<Element>
init(queueSize: Int) {
self._queue = Queue(capacity: queueSize + 1)
}
override func addValueToBuffer(_ value: Element) {
self._queue.enqueue(value)
}
override func replayBuffer<O: ObserverType>(_ observer: O) where O.E == Element {
for item in self._queue {
observer.on(.next(item))
}
}
override func _synchronized_dispose() {
super._synchronized_dispose()
self._queue = Queue(capacity: 0)
}
}
AsyncSubject
AsyncSubject只发送最后一个事件，并且只在Observable发送complete之后。如果被error终端发送，则不会发送任何事件，而是会发送这个错误事件。
直接上代码: 如果调用了onCompleted,那么会打印订阅到了:4，如果不注释onError,那么会打印订阅到了: error(Error Domain=test Code=10086 "(null)").
// 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: "test", code: 10086, userInfo: nil))
asynSub.onCompleted()
总结原理(自行探索）: 保存最后一次事件lastElement,只有在.error或.next的时候才发送出去。
Variable : 5.0已经废弃
使用快速方法直接赋值来发送事件，也可以直接用value属性直接取值。
以下代码打印的是:"订阅到了: next(10) 订阅到了: next(1000) 订阅到了: completed"
// 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,只不过封装了一层而已，能够快速赋值。（由于不符合RxSwift的编程习惯，已经废弃，不做重点）
private let _subject: BehaviorSubject<Element>
BehaviorRelay: 替代 Variable
直接上代码，它是在swift5.0一般用来替代Variable的，只是赋值的方法由直接给value属性赋值修改为了使用accept赋值，使用起来一样的通过value可以直接获取到值。
let behaviorRelay = BehaviorRelay.init(value: 100)
print("值:\(behaviorRelay.value)")
behaviorRelay.accept(20);
print("值:\(behaviorRelay.value)")
总结
以上就是关于RxSwift里的Subject的总结，在开发中使用的比较多，所以原理和使用方法是一定要掌握的。
