RxSwift之深入解析Subject的使用和实现原理
【摘要】
一、Subject
RxSwift 的核心逻辑 Observable 不具备发送事件的能力，创建一个 Observable 的时候就要预先将要发出的数据都准备好，等到有人订阅它时再将数据通过 Event...
一、Subject
- RxSwift 的核心逻辑 Observable 不具备发送事件的能力，创建一个 Observable 的时候就要预先将要发出的数据都准备好，等到有人订阅它时再将数据通过 Event 发出去。但有时希望 Observable 在运行时能动态地获得或者说产生一个新的数据，再通过 Event 发送出去。比如，订阅一个输入框的输入内容，当用户每输入一个字符之后，输入框关联的 Observable 就会发出一个带有输入内容的 Event，通知给所有订阅者。为此，RxSwift 提供了一种可以发送事件又可以订阅事件值的对象，它就是 Subject。
- Subject 既是订阅者，也是 Observable：它是订阅者，是因为能够动态地接收新的值；它是 Observable，是因为当 Subjects 有了新的值之后，就会通过 Event 将新值发出给它的所有订阅者。
- Subject 常用的方法：
-
- onNext( : )：是 on(.next( : )) 的简写，该方法相当于 subject 接收到一个.next 事件；
-
- onError( : )：是 on(.error( : )) 的简写，该方法相当于 subject 接收到一个 .error 事件；
-
- onCompleted()：是 on(.completed) 的简写，该方法相当于 subject 接收到一个 .completed 事件。
二、PublishSubject
- PublishSubject 不需要初始值就能创建，它的订阅者从开始订阅的时间点起，可以收到订阅后 Subject 发出的新 Event，而不会收到它们在订阅前已发出的 Event，即 PublishSubject 仅仅发送在订阅之后由源 Observable 发送的数据。
- 如下所示，最上面是 PublishSubject，下面分别表示两个新的订阅，它们订阅的时间点不同，可以发现 PublishSubject 的订阅者只能收到它们订阅后的 Event：
- PublishSubject 一旦被建立就会立刻开始发送事件（除非采取方法去阻止它），这种机制有丢失事件的风险，因为在 Subject 被创建和被监听之间有一定的时间间隔，如果想保证所有的事件都可以被监听到的话，可以有两种方法：
-
- 第一种方法是使用 Create 方法（在发送之前检查是否所有 observer 已经订阅）；
-
- 第二种方法是可以使用 ReplaySubject。
- 如果源 Observable 被一个 error 中断，PublishSubject 将不会发送任何事件给后续的 observer，但是它会传递 error 信息，如下所示：
- 使用示例如下：
let disposeBag = DisposeBag()
// 创建一个PublishSubject
let subject = PublishSubject<String>()
// 由于当前没有任何订阅者，所以该信息不会被输出到控制台
subject.onNext("1")
// 第1次订阅subject
subject.subscribe(onNext: { string in
print("第1次订阅：", string)
}, onCompleted:{
print("第1次订阅：onCompleted")
}).disposed(by: disposeBag)
// 当前有1个订阅，则该信息会被输出到控制台
subject.onNext("2")
// 第2次订阅subject
subject.subscribe(onNext: { string in
print("第2次订阅：", string)
}, onCompleted:{
print("第2次订阅：onCompleted")
}).disposed(by: disposeBag)
// 当前有2个订阅，则该信息会输出到控制台
subject.onNext("3")
// subject结束
subject.onCompleted()
// subject完成后会发出.next事件
subject.onNext("4")
// subject完成后它的所有订阅（包括结束后的订阅），都能收到subject的.completed事件
subject.subscribe(onNext: { string in
print("第3次订阅：", string)
}, onCompleted:{
print("第3次订阅：onCompleted")
}).disposed(by: disposeBag)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 运行结果如下：
第1次订阅： 2
第1次订阅： 3
第2次订阅： 3
第1次订阅：onCompleted
第2次订阅：onCompleted
第3次订阅：onCompleted
- 1
- 2
- 3
- 4
- 5
- 6
三、AsyncSubject
- AsyncSubject 只发送由源 Observable 发送的最后一个事件，并且只在源 Observable 完成之后（如果源 Observable 没有发送任何值，AsyncSubject 也不会发送任何值）：
- AsyncSubject 会发送相同的值给所有 observer。但是，如果源 Observable 被 error 中断了发送，AsyncSubject 便不会发送任何事件，而是会发送从源 Observable 传来的 error 提示：
- 使用示例：
let disposeBag = DisposeBag()
// 创建序列
let asynSub = AsyncSubject<Int>.init()
// 发送信号
asynSub.onNext(1)
asynSub.onNext(2)
// 订阅序列
asynSub.subscribe{ print("订阅到:",$0)}
.disposed(by: disposeBag)
// 再次发送
asynSub.onNext(3)
asynSub.onNext(4)
// asynSub.onError(NSError.init(domain: "dw", code: 10086, userInfo: nil))
asynSub.onCompleted()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 运行结果如下：
订阅到: next(4)
订阅到: completed
- 1
- 2
四、BehaviorSubject
- 当一个 observer 订阅一个 BehaviorSubject，它就开始发送最近由源 Observable 发送的事件（或者是还没有被发送的种子值/默认值），然后继续发送从源 Observable 接收到的其它事件。如下所示：
- 如果源 Observable 被一个 error 中断，那么 BehaviorSubject 不会发送事件给后续的 observer，但会传递给它们 error 的信息。如下所示：
- 使用示例：
let disposeBag = DisposeBag()
// 创建一个BehaviorSubject
let subject = BehaviorSubject(value: "1")
// 第1次订阅subject
subject.subscribe { event in
print("第1次订阅：", event)
}.disposed(by: disposeBag)
// 发送next事件
subject.onNext("2")
// 发送error事件
subject.onError(NSError(domain: "dw", code: 0, userInfo: nil))
// 第2次订阅subject
subject.subscribe { event in
print("第2次订阅：", event)
}.disposed(by: disposeBag)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 运行结果如下：
第1次订阅： next(1)
第1次订阅： next(2)
第1次订阅： error(Error Domain=dw Code=0 "(null)")
第2次订阅： error(Error Domain=dw Code=0 "(null)")
- 1
- 2
- 3
- 4
五、ReplaySubject
- ReplaySubject 在创建时候需要设置一个 bufferSize，表示它对于它发送过的 event 的缓存个数。比如一个 ReplaySubject 的 bufferSize 设置为 2，它发出了 3 个 .next 的 event，那么它会将后两个（最近的两个）event 给缓存起来。此时如果有一个 subscriber 订阅了这个 ReplaySubject，那么这个 subscriber 就会立即收到前面缓存的两个.next 的 event。
- 如果一个 subscriber 订阅已经结束的 ReplaySubject，除了会收到缓存的 .next 的 event外，还会收到那个终结的 .error 或者 .complete 的event。
- 如下所示，最上面是 ReplaySubject（bufferSize 设为为 2），下面分别表示两个新的订阅，它们订阅的时间点不同，可以发现 ReplaySubject 的订阅者一开始就能收到 ReplaySubject 之前发出的两个 Event（如果有）：
- 使用示例：
let disposeBag = DisposeBag()
// 创建一个bufferSize为2的ReplaySubject
let subject = ReplaySubject<String>.create(bufferSize: 2)
// 连续发送3个next事件
subject.onNext("1")
subject.onNext("2")
subject.onNext("3")
// 第1次订阅subject
subject.subscribe { event in
print("第1次订阅：", event)
}.disposed(by: disposeBag)
// 再发送1个next事件
subject.onNext("4")
// 第2次订阅subject
subject.subscribe { event in
print("第2次订阅：", event)
}.disposed(by: disposeBag)
// 让subject结束
subject.onCompleted()
// 第3次订阅subject
subject.subscribe { event in
print("第3次订阅：", event)
}.disposed(by: disposeBag)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 运行结果如下：
第1次订阅： next(2)
第1次订阅： next(3)
第1次订阅： next(4)
第2次订阅： next(3)
第2次订阅： next(4)
第1次订阅： completed
第2次订阅： completed
第3次订阅： next(3)
第3次订阅： next(4)
第3次订阅： completed
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
六、Variable
- Variable 其实就是对 BehaviorSubject 的封装，所以它也必须要通过一个默认的初始值进行创建，它具有 BehaviorSubject 的功能，能够向它的订阅者发出上一个 event 以及之后新创建的 event。
- 不同的是，Variable 还会把当前发出的值保存为自己的状态，同时它会在销毁时自动发送 .complete 的 event，不需要也不能手动给 Variables 发送 completed 或者 error 事件来结束它。
- 简单地说，就是 Variable 有一个 value 属性，改变这个 value 属性的值就相当于调用一般 Subjects 的 onNext() 方法，而这个最新的 onNext() 的值就被保存在 value 属性里，直到再次修改它。
- Variables 本身没有 subscribe() 方法，但是所有 Subjects 都有一个 asObservable() 方法，可以使用这个方法返回这个 Variable 的 Observable 类型，拿到这个 Observable 类型就能订阅它。
- 使用示例：
let disposeBag = DisposeBag()
// 创建一个初始值为1的Variable
let variable = Variable("1")
// 修改value值
variable.value = "2"
// 第1次订阅
variable.asObservable().subscribe {
print("第1次订阅：", $0)
}.disposed(by: disposeBag)
// 修改value值
variable.value = "3"
// 第2次订阅
variable.asObservable().subscribe {
print("第2次订阅：", $0)
}.disposed(by: disposeBag)
// 修改value值
variable.value = "4"
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 运行结果：
第1次订阅： next(2)
第1次订阅： next(3)
第2次订阅： next(3)
第1次订阅： next(4)
第2次订阅： next(4)
第1次订阅： completed
第2次订阅： completed
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- Variable 虽然被废弃了，但是由于 Variable 的灵活性，因此在开发里面应用非常之多。
七、BehaviorRelay
- BehaviorRelay 会替换原来的 Variable，储存一个信号，并且可以随时订阅响应，但响应发送的时候要注意 behaviorR.accept(20)：
let disposeBag = DisposeBag()
let behaviorRelay = BehaviorRelay(value: 100)
behaviorRelay.subscribe(onNext: { (num) in
print(num)
.disposed(by: disposeBag)
print("打印:\(behaviorRelay.value)")
behaviorRelay.accept(1000)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
八、Subject 原理分析
① SubjectType
- SubjectType 继承自 ObservableType，具有序列特性，并且关联观察者类型，具备观察者类型的能力，如下：
public protocol SubjectType : ObservableType {
// 关联观察者类型，具备观察者类型的能力
associatedtype SubjectObserverType : ObserverType
func asObserver() -> SubjectObserverType
}
- 1
- 2
- 3
- 4
- 5
- 现有如下实例 subject：
let disposeBag = DisposeBag()
// 初始化序列
let publishSub = PublishSubject<Int>()
// 发送响应序列
publishSub.onNext(1)
// 订阅序列
publishSub.subscribe { print("订阅到:",$0)}
.disposed(by: disposeBag)
// 再次发送响应
publishSub.onNext(2)
publishSub.onNext(3)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- PublishSubject 很明显能够订阅信号（序列最基本的能力），并且能够发送响应，又是观察者的能力。
② 订阅响应流程
public override func subscribe -> Disposable {
self._lock.lock()
let subscription = self._synchronized_subscribe(observer)
self._lock.unlock()
return subscription
}
func _synchronized_subscribe -> Disposable {
......
......
let key = self._observers.insert(observer.on)
return SubscriptionDisposable(owner: self, key: key)
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- self._observers.insert(observer.on): 通过一个集合添加进去所有的订阅事件，很明显在合适的地方一次性全部执行。其中也返回这次订阅的销毁者，方便执行之后的工作：synchronizedUnsubscribe->self._observers.removeKey(disposeKey)。
mutating func removeKey(_ key: BagKey) -> T? {
if _key0 == key {
_key0 = nil
let value = _value0!
_value0 = nil
return value
}
if let existingObject = _dictionary?.removeValue(forKey: key) {
return existingObject
}
for i in 0 ..< _pairs.count where _pairs[i].key == key {
let value = _pairs[i].value
_pairs.remove(at: i)
return value
}
return nil
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 遍历通过 key 获取响应 bag 中的 value，执行集合移除，因为没有相应持有关系，达到自动释放销毁。
③ 发送信号流程
public func on(_ event: Event<Element>) {
dispatch(self._synchronized_on(event), event)
}
- 1
- 2
- 3
- 这里调用了 dispatch 函数，传了两个参数：self._synchronized_on(event) 和 event，查看 dispatch 函数源码：
func dispatch<E>(_ bag: Bag) {
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)
}
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- bag._value0?(event) 首先执行事件的回调，然后判断 bag._onlyFastPath 的情况，默认会开启快速通道，如果是开启慢速通道，需要从刚刚添加进 bag 包里面的匹配，挨个进行 pairs[i].value(event) 外界事件回调，然后拿回外界封装的闭包的闭包调用 :element(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()
}
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 如果 self._isDisposed || self._stopped 成立，就会返回一个空的集合，也就没有序列的响应。在 .completed, .error 都会改变状态 self._stopped = true，也就是说序列完成或者错误之后都无法再次响应。在.completed, .error 还会移除添加在集合里面的内容。
- Subject 流程图如下（Subject 把订阅流程和响应流程都内部实现，所以没有必要引入 sink）：
④ Subject 对比
- PublishSubject、BehaviorSubject、ReplaySubject、AsyncSubject、Variable，它们之间既有各自的特点，也有相同之处：
-
- 首先它们都是 Observable，它们的订阅者都能收到它们发出的新的 Event；
-
- 直到 Subject 发出 .complete 或者 .error 的 Event 后，该 Subject 便终结，同时它也就不会再发出 .next 事件；
-
- 对于那些在 Subject 终结后再订阅它的订阅者，也能收到 subject 发出的一条 .complete 或 .error 的 event，告诉新的订阅者它已经终结。
- 它们之间最大的区别只是在于：当一个新的订阅者刚订阅它的时候，能不能收到 Subject 以前发出过的旧 Event，如果能的话又能收到多少个 Event。
文章来源: blog.csdn.net，作者：Serendipity·y，版权归原作者所有，如需转载，请联系作者。
原文链接：blog.csdn.net/Forever_wj/article/details/120999314
【版权声明】本文为华为云社区用户转载文章，如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱：
cloudbbs@huaweicloud.com
- 点赞
- 收藏
- 关注作者
评论（0）