RxSwift调度者-scheduler源码解析（下）

这一篇关于调度的具体流程分析，如有对RxSwift调度者不太了解的观客，请移步RxSwift调度者-scheduler源码解析（上）

observeOn&subscribeOn

上面两个函数是 RxSwift 切换调度环境非常重要的，可以说掌握了这两个函数对 RxSwift调度者 的理解可谓是：指日可待

首先我给大家带来一个测试：

DispatchQueue.global().async {
    self.actionBtn.rx.tap
        .subscribe(onNext: { () in
            print("点击了按钮 --- \(Thread.current)")
        })
        .disposed(by: self.bag)
}

• 问：当前按钮点击，打印的线程情况是什么：
  点击了按钮 --- <NSThread: 0x600000c2d980>{number = 1, name = main}
• 你是否回答正确？这里切换到了主线程，那么我明明规定了当前线程就是全局并发队列（子线程），但是为什么切换过来的？

image

public func controlEvent(_ controlEvents: UIControl.Event) -> ControlEvent<()> {
    let source: Observable<Void> = Observable.create { [weak control = self.base] observer in
            // 调度主线程判断
            MainScheduler.ensureRunningOnMainThread()
        }
    return ControlEvent(events: source)
}

• 我们一般思维跟进：controlEvent,发现我们调度执行必须要在主队列！
• NO 现在我不要知道这个，我想知道它是怎么切过去的！

public init<Ev: ObservableType>(events: Ev) where Ev.Element == Element {
    self._events = events.subscribeOn(ConcurrentMainScheduler.instance)
}

public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable {
    return self._events.subscribe(observer)
}

• OK 很明显我们的 ControlEvent 的序列 subscribe 是调用了一个函数就是：subscribeOn.
• 其中ConcurrentMainScheduler.instance 内部封装了 主队列！那么所有事情也就清晰了！
• 哈哈，其实还有一个地方还是不明确的就是：subscribeOn 到底内部的逻辑又是什么？请看下面的分析

public func subscribeOn(_ scheduler: ImmediateSchedulerType)
    -> Observable<Element> {
    return SubscribeOn(source: self, scheduler: scheduler)
}

• 看到返回值的类型就知道，原来的序列是被subscribeOn进行处理了，封装了中间层：SubscribeOn 的序列

final private class SubscribeOn<Ob: ObservableType>: Producer<Ob.Element> {
    let source: Ob
    let scheduler: ImmediateSchedulerType
    
    init(source: Ob, scheduler: ImmediateSchedulerType) {
        self.source = source
        self.scheduler = scheduler
    }
    
    override func run(_ observer: cancel:) -> (sink:subscription:) {
        let sink = SubscribeOnSink(parent: self, observer: observer, cancel: cancel)
        let subscription = sink.run()
        return (sink: sink, subscription: subscription)
    }
}

• 看到 SubscribeOn 的继承关系（Producer）,我们对他也就放心了。
• 序列订阅的时候，会创建一个observer的观察者
• 经过Producer 流回SubscribeOn的run
• 在经过 SubscribeOnSink.run 到观察者的回调（或者内部源序列的订阅，传sink作为观察者回调，后面的流程只是重复走了一次）
• 由观察者的发送响应，回到 sink 的 on
• 由 sink的属性观察者（也就是中间封装保存的）响应event事件
• 最后调用外界的subscriber的闭包
• 下面我们直接跳过上面的流程，进入关于调度相关代码（这里因为篇幅原因，我也简写流程，看核心源码）

func run() -> Disposable {
    let disposeEverything = SerialDisposable()
    let cancelSchedule = SingleAssignmentDisposable()
    
    disposeEverything.disposable = cancelSchedule
    
    let disposeSchedule = self.parent.scheduler.schedule(()) {

        let subscription = self.parent.source.subscribe(self)
        disposeEverything.disposable = ScheduledDisposable(scheduler: disposable:)
        return Disposables.create()
    }
    cancelSchedule.setDisposable(disposeSchedule)
    return disposeEverything
}

• 这里就有一个非常重要的方法：self.parent.scheduler.schedule(),背后的逻辑我们在前面的代码也解析过了！
• 调用self.scheduleInternal(state, action: action)

func schedule<StateType>(_ state: action: ) -> Disposable {
    let cancel = SingleAssignmentDisposable()
    self.queue.async {
        if cancel.isDisposed {
            return
        }
        cancel.setDisposable(action(state))
    }
    return cancel
}

• 看到这里你也就明白了我们调度的是把任务 action封装，利用 scheduler 统一管理
• 很多同学如果不求甚解，那么到这里也差不多了！不怕死的同学可以继续跟着我往下面走
• 其实这里的action就是一个 schduler 调用时候的闭包，就会执行：let subscription = self.parent.source.subscribe(self), 这里距离的我们的目标还有一点距离，哈哈哈哈
• 源序列的subscribe,必然会来到Producer

override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
    if !CurrentThreadScheduler.isScheduleRequired {
        // The returned disposable needs to release all references once it was disposed.
        let disposer = SinkDisposer()
        let sinkAndSubscription = self.run(observer, cancel: disposer)
        disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

        return disposer
    }
    else {
        return CurrentThreadScheduler.instance.schedule(()) { _ in
            let disposer = SinkDisposer()
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
    }
}

• 这里会根据当前的调度环境来判断，其实内部的流程差不多，只是多了一个方法调用

public func schedule<StateType>(_ state: action: ) -> Disposable {

    if CurrentThreadScheduler.isScheduleRequired {
      // 已经标记，就置false
        CurrentThreadScheduler.isScheduleRequired = false
     // 外界闭包调用执行
        let disposable = action(state)
      // 延迟销毁 
        defer {
            CurrentThreadScheduler.isScheduleRequired = true
            CurrentThreadScheduler.queue = nil
        }
      // 先省略。。。
        return disposable
    }
     // 先省略。。。
    return scheduledItem
}

• 由这里就可以看出直接调度执行，但是问题在哪呢？
  • 如果你当前调度环境不变，那就没有问题！一直都在我贴出的代码流程来回执行
  • 如果我这里调度的是子线程，那么就完全不一样，针对当前队列，还有线程安全都是需要处理的！

public func scheduleRecursive<State>(_ state: action: ) -> Disposable {
    // 递归调度者
    let recursiveScheduler = RecursiveImmediateScheduler(action: scheduler:)
    // 调度状态执行
    recursiveScheduler.schedule(state)
    return Disposables.create(with: recursiveScheduler.dispose)
}

• 在这个递归调度者里面的逻辑就相对来说，比较复杂！请认真跟我分析

func schedule(_ state: State) {
    var scheduleState: ScheduleState = .initial
    let d = self._scheduler.schedule(state) { state -> Disposable in     
        // 这里因为在递归环境，加了一把锁递归锁，保障安全   
        let action = self._lock.calculateLocked { () -> Action? in
                 return self._action
        }
        
        if let action = action {
            action(state, self.schedule)
        }
        
        return Disposables.create()
    }
  // 篇幅，先省略，大家自行查阅  
}

• 这里因为在递归环境，加了一把锁递归锁，保障安全
• 通过保护，获取action执行,也就是外界传给递归调度者的闭包任务
• 看到这里，是不是你已经get 到了为什么RxSwift 的数组调度出来是有顺序的，因为在递归调度，已经加锁了，保障线程资源安全
• 执行完源序列的响应，会吧任务保存进队列，这里还是需要讲解的！

public func schedule<StateType>(_ state: StateType, action: ) -> Disposable {
    // 上面的流程就省略了
    let existingQueue = CurrentThreadScheduler.queue

    let queue: RxMutableBox<Queue<ScheduledItemType>>
    if let existingQueue = existingQueue {
        queue = existingQueue
    }
    else {
        queue = RxMutableBox(Queue<ScheduledItemType>(capacity: 1))
        CurrentThreadScheduler.queue = queue
    }

    let scheduledItem = ScheduledItem(action: action, state: state)
    queue.value.enqueue(scheduledItem)

    return scheduledItem
}

• 把任务和状态封装成了ScheduledItem,面向对象，更容易传输&执行
• 把这个事务queue.value.enqueue(scheduledItem),排队进队列

public func schedule<StateType>(_ state: StateType, action: @escaping (StateType) -> Disposable) -> Disposable {
    if CurrentThreadScheduler.isScheduleRequired {

        CurrentThreadScheduler.isScheduleRequired = false
        let disposable = action(state)  

         // 判断当前队列情况，是否存在
        guard let queue = CurrentThreadScheduler.queue else {
            return disposable
        }
        // 从队列去除任务
        while let latest = queue.value.dequeue() {
            if latest.isDisposed {
                continue
            }
            latest.invoke()
        }

        return disposable
    }
    /// 省略
}

• 流程任务执行action(state) 完毕之后，又会执行下面的流程
• 判断当前队列情况，是否存在
• 从队列去除任务 ： queue.value.dequeue()
• 重点：latest.invoke()

func invoke() {
     self._disposable.setDisposable(self._action(self._state))
}

• 就是原来响应回来时候保存的 action执行
• 只不过加了销毁的机制
• 这个时候我们的流程就会由原来的 源序列 流进 ObserveOnSink
• 保障了在 ObserveOnSink 的调度环境是有序的进队的：self._queue.enqueue(event)
• 执行self._scheduler.scheduleRecursive((), action: self.run)

override func onCore(_ event: Event<Element>) {
    let shouldStart = self._lock.calculateLocked { () -> Bool in
        self._queue.enqueue(event)
    }
    if shouldStart {
        self._scheduleDisposable.disposable = 
        self._scheduler.scheduleRecursive((), action: self.run)
    }
}

• 这里的手法是非常重要的：毕竟并发队列很可能存在！我们模拟的也是并发

func run(_ state: (), _ recurse: (()) -> Void) {
    // 加锁获取观察者，很队列任务
    let (nextEvent, observer) = self._lock.calculateLocked { 
        if !self._queue.isEmpty {
            return (self._queue.dequeue(), self._observer)
        }
    }
   
    // 观察者发送响应
    if let nextEvent = nextEvent, !self._cancel.isDisposed {
        observer.on(nextEvent)
        if nextEvent.isStopEvent {
            self.dispose()
        }
    }
}

• 加锁获取观察者，很队列任务 : (self._queue.dequeue(), self._observer)
• 观察者发送响应: observer.on(nextEvent)
• 这个时候我们外界的订阅闭包就可以如期回调咯！

完美！解析到这里，整个流程解析完毕，就问此时此刻还有谁？
45度仰望天空，该死！我这无处安放的魅力！

总结

整个流程是比较复杂，其实如果你这个时候，整体看源码（上帝的视角）！不难得出：

• 源序列包装
• 内部序列创建
• 调度环境&观察者传递准备
• 源序列订阅 - 根据调度环境调度 - 流程流到观察者就是我们中间内部序列的Sink
• Sink 调度执行 响应发给观察者
• 由观察者响应 订阅事件event

就是两层序列订阅响应，我的第二层的 sink 就是源序列的观察者

RxSwift的源码其实你只要掌握我前面的分享过的核心逻辑，后面的内容都是一样的，只不过嵌套，胶水代码依赖建立。其中依赖关系，中间层封装，流程走位是各位在接下来的iOS进阶非常重要的思维，告别垃圾代码，迎接未来!我们一起加油：和谐学习，不急不躁