RxSwift(九)Subjects使用总结及原理解析

作者: 伤心的EasyMan | 来源:发表于2019-08-12 01:29 被阅读0次

前言

RxSwift中的Subjects非常特殊，因为它即充当了观察者序列(Observable)的角色，还充当了观察者(Observer)的角色。
本文的探索比较简单，如果之前有探索过RxSwift核心逻辑的话，那再看本文是相当容易的。

PublishSubject

PublishSubject:一开始的时候是空的，只发送新的元素给观察者。它仅仅会发送observer订阅之后的事件，也就是说如果sequence上有.next的到来，但是这个时候某个observer还没有subscribe它，这个observer就收不到这条信息，它只会收到它订阅之后发生的事件。

image
简单来说:就是订阅后才接收发出的新的事件，之前的统统不管。

如果不理解，直接上代码

     // 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的总结，在开发中使用的比较多，所以原理和使用方法是一定要掌握的。

RxSwift(九)Subjects使用总结及原理解析

前言

PublishSubject

BehaviorSubject

ReplaySubject

AsyncSubject

Variable : 5.0已经废弃

BehaviorRelay: 替代 Variable

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读