调度和 _synchronized_subscribe RxSwift

Question

我想了解 RxSwift 的核心逻辑，我有一个简单的按钮，当 onNext(5) 触发时，它会触发并打印控制台。

我正在关注 Xcode 左侧的话题，如下所示。

-RxSwift 库

// Step 1
  public func onNext(_ element: E) {
        self.on(.next(element))
    }
// Step 2
    public func on(_ event: Event<Element>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        dispatch(self._synchronized_on(event), event)
    }
//Step 3
@inline(__always)
func dispatch<E>(_ bag: Bag<(Event<E>) -> Void>, _ event: Event<E>) {
    bag._value0?(event) 
/// after dispatch calling, it is creating BagKey as below
    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) // directly flow comes here and my question is related with this step
        return SubscriptionDisposable(owner: self, key: key)
    }
// Step 4
        MapSink.on(_:)
// Step 5 FilterSink - PublishSubject... and flow goes on prints

我的代码

import UIKit
import RxSwift

  class ViewController: UIViewController {

      let publisher = PublishSubject<Int>()
  let disposeBag = DisposeBag()

  @IBOutlet weak var button: UIButton!

  override func viewDidLoad() {
      super.viewDidLoad()

      publisher
              .map { [=12=] * 2}
              .filter { [=12=] > 4 }
              .subscribe(onNext: { n in
                  print("We've got number \(n)!")
              })
              .disposed(by: disposeBag)
  }

  @IBAction func trigger(_ sender: Any) {
      print("button triggered")
      publisher.onNext(5)
  }
 }

我的问题是第 3 步发生了什么，你能解释一下吗？我也查看了 synchronized_subscribe<O: ObserverType> 函数调用的地方，但没有找到；线程直接创建一个键，来到这个函数里面。为什么 RxSwift 会跳到那里，即使它没有被调用，这是什么魔法？

Answer 1

我不确定这里真正问的是什么......逻辑是当调用 onNext(5) 时，它会向它的每个订阅者发送一个 Event.next(5)（其中有只有一个。）

至于调用了什么函数，链条是这样的：

ObserverType.onNext(_ element: Element)
    PublishSubject<Element>.on(_ event: Event<Element>)
        PublishSubject<Element>.synchronized_on(_ event: Event<Element>) // gets Bag of observers
        dispatch<Element>(_ bag: Bag<(Event<Element>) -> Void>, _ event: Event<Element>) // dispatches event to each observer in the bag.
            MapSink<SourceType, Observer: ObserverType>.on(_ event: Event<Self.Element>) // this is one of the observers.

您似乎对 dispatch(_:_:) 函数感到困惑。这个函数接受一个 Bag of observers（严格来说，一个接受 Events 的闭包容器）并将事件发送到 Bag 中包含的一个 Observer。

调度和 _synchronized_subscribe RxSwift

dispatch and _synchronized_subscribe RxSwift

swift

rx-swift