RxSwift源码分析(五)-调度器Scheduler

在RxSwift中主要有如下四个成员：

• 可观察序列 - Observable
• 观察者 - Observer
• 调度者 - Scheduler
• 销毁者 - Dispose

如果这四个都弄明白了，那么可以说整个RxSwift也就弄明白了。这篇文章来具体分析调度者 - Scheduler

什么是调度者

Schedulers 是 Rx 实现多线程的核心模块，它主要用于控制任务在哪个线程或队列运行，它内部的实现是对GCD和OperationQueue进行了封装。
如果你曾经使用过 GCD， 那你对以下代码应该不会陌生，功能都是从多线程获取数据然后到主线程刷新UI：

// 后台取得数据，主线程处理结果
DispatchQueue.global(qos: .userInitiated).async {
    let data = try? Data(contentsOf: url)
    DispatchQueue.main.async {
        self.data = data
    }
}

如果用 RxSwift 来实现，大致是这样的：

let rxData: Observable<Data> = ...

rxData
    .subscribeOn(ConcurrentDispatchQueueScheduler(qos: .userInitiated))
    .observeOn(MainScheduler.instance)
    .subscribe(onNext: { [weak self] data in
        self?.data = data
    })
    .disposed(by: disposeBag)

感受了Scheduler的使用之后，来看看里面具体是如何实现的。

RxSwift中的几种Scheduler的介绍

CurrentThreadScheduler

• 表示当前线程，默认就在当前线程上。

public class CurrentThreadScheduler : ImmediateSchedulerType {
    typealias ScheduleQueue = RxMutableBox<Queue<ScheduledItemType>>

    /// The singleton instance of the current thread scheduler.
    public static let instance = CurrentThreadScheduler()

    private static var isScheduleRequiredKey: pthread_key_t = { () -> pthread_key_t in
        let key = UnsafeMutablePointer<pthread_key_t>.allocate(capacity: 1)
        defer { key.deallocate() }
                                                               
        guard pthread_key_create(key, nil) == 0 else {
            rxFatalError("isScheduleRequired key creation failed")
        }

        return key.pointee
    }()

    private static var scheduleInProgressSentinel: UnsafeRawPointer = { () -> UnsafeRawPointer in
        return UnsafeRawPointer(UnsafeMutablePointer<Int>.allocate(capacity: 1))
    }()

    static var queue : ScheduleQueue? {
        get {
            return Thread.getThreadLocalStorageValueForKey(CurrentThreadSchedulerQueueKey.instance)
        }
        set {
            Thread.setThreadLocalStorageValue(newValue, forKey: CurrentThreadSchedulerQueueKey.instance)
        }
    }

    /// Gets a value that indicates whether the caller must call a `schedule` method.
    public static fileprivate(set) var isScheduleRequired: Bool {
        get {
            return pthread_getspecific(CurrentThreadScheduler.isScheduleRequiredKey) == nil
        }
        set(isScheduleRequired) {
            if pthread_setspecific(CurrentThreadScheduler.isScheduleRequiredKey, isScheduleRequired ? nil : scheduleInProgressSentinel) != 0 {
                rxFatalError("pthread_setspecific failed")
            }
        }
    }
    ......
}

• isScheduleRequired 用来表示是否必须调用schedule方法,利用对 queue的set,get方法的观察，绑定我们的当前队列与静态字符串，实现同一线程数据共享。

SerialDispatchQueueScheduler

• SerialDispatchQueueScheduler 抽象了串行 DispatchQueue。如果你需要执行一些串行任务，可以切换到这个 Scheduler 运行。

ConcurrentDispatchQueueScheduler

• ConcurrentDispatchQueueScheduler 抽象了并行 DispatchQueue。如果你需要执行一些并发任务，可以切换到这个 Scheduler 运行。

OperationQueueScheduler

• OperationQueueScheduler 抽象了 NSOperationQueue。它具备 NSOperationQueue 的一些特点，例如，你可以通过设置 maxConcurrentOperationCount，来控制同时执行并发任务的最大数量。

public class OperationQueueScheduler: ImmediateSchedulerType {
    public let operationQueue: OperationQueue
    public let queuePriority: Operation.QueuePriority
    
    /// Constructs new instance of `OperationQueueScheduler` that performs work on `operationQueue`.
    ///
    /// - parameter operationQueue: Operation queue targeted to perform work on.
    /// - parameter queuePriority: Queue priority which will be assigned to new operations.
    public init(operationQueue: OperationQueue, queuePriority: Operation.QueuePriority = .normal) {
        self.operationQueue = operationQueue
        self.queuePriority = queuePriority
    }
    ......
}

• 在初始化 OperationQueueScheduler 对象时，需要传入 OperationQueue 和 优先级queuePriority，作为初始化参数。

MainScheduler

• MainScheduler 代表主线程。如果你需要执行一些和 UI 相关的任务，就需要切换到该 Scheduler 运行。

public final class MainScheduler : SerialDispatchQueueScheduler {

    private let _mainQueue: DispatchQueue

    let numberEnqueued = AtomicInt(0)
    public init() {
        self._mainQueue = DispatchQueue.main
        super.init(serialQueue: self._mainQueue)
    }
    public static let instance = MainScheduler()
}

• 通过源码可以看出MainScheduler继承了SerialDispatchQueueScheduler串行队列，因为主队列本来就是一个特殊的串行队列。然后在初始化对象时，确定了队列类型为主队列self._mainQueue = DispatchQueue.main。

使用

根据前面的示例来分析下subscribeOn和observeOn的具体实现
使用 subscribeOn

• 我们用 subscribeOn 来决定数据序列的构建函数在哪个 Scheduler 上运行。以上例子中，由于获取 Data 需要花很长的时间，所以用 subscribeOn 切换到 后台 Scheduler 来获取 Data。这样可以避免主线程被阻塞。
• 默认情况下，Observable 创建，应用操作符以及发出通知都会在 Subscribe 方法调用的 Scheduler 执行。subscribeOn 操作符将改变这种行为，它会指定一个不同的 Scheduler 来让 Observable 执行。
• 源码分析：

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

• 看到返回值类型是Observable，我们就知道原来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: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Ob.Element {
        let sink = SubscribeOnSink(parent: self, observer: observer, cancel: cancel)
        let subscription = sink.run()
        return (sink: sink, subscription: subscription)
    }
}

• 在初始化的时候保存了源序列和调度者。
• 通过之前对RxSwift核心逻辑的讲解的文章，就能够知道，当序列被订阅的时候，代码一定会执行到run方法来。（还不太了解的朋友可以查看我前面的关于RxSwift核心逻辑的文章）
  进入到SubscribeOnSink.run方法

func run() -> Disposable {
    let disposeEverything = SerialDisposable()
    let cancelSchedule = SingleAssignmentDisposable()
    
    disposeEverything.disposable = cancelSchedule
    
    let disposeSchedule = self.parent.scheduler.schedule(()) { _ -> Disposable in
        let subscription = self.parent.source.subscribe(self)
        disposeEverything.disposable = ScheduledDisposable(scheduler: self.parent.scheduler, disposable: subscription)
        return Disposables.create()
    }

    cancelSchedule.setDisposable(disposeSchedule)

    return disposeEverything
}

• 看到有一句关于调度相关的代码self.parent.scheduler.schedule()，self.parent.scheduler就是调用SubscribeOn方法作为参数传进来的队列，然后执行schedule方法。
• 调用self.scheduleInternal(state, action: action)
• 然后执行到self.configuration.schedule(state, action: action)

func schedule<StateType>(_ state: StateType, action: @escaping (StateType) -> Disposable) -> Disposable {
    let cancel = SingleAssignmentDisposable()

    self.queue.async {
        if cancel.isDisposed {
            return
        }
        cancel.setDisposable(action(state))
    }

    return cancel
}

• 看到这里终于明白了，原来就是把任务放在了设置的队列里面异步执行
• 这个action(state)就是从外面传进来的尾随闭包，所以代码会开始执行闭包，就会执行let subscription = self.parent.source.subscribe(self)，对源序列进行订阅，所以必然会来到Producer的subscribe方法。

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
        }
    }
}

• 这里会根据当前的调度环境来判断具体执行哪块代码，这里首先会走else里面。来到schedule方法里面
• defer是延迟调用，保证在return之前调用
• 首先会执行action(state)，在闭包里面会执行ObservableSequenceSink.run方法，最后代码又会来到这个schedule方法，由于上一次进来时把isScheduleRequired设置成了false，所以代码会执行代码块3(如图)
• 代码块3主要是将任务封装成一个ScheduledItem对象，然后加入到队列中
• 然后执行代码块2，代码块2的作用是把前面存进队列的任务一个个拿出来按顺序执行，即开始执行下面的闭包

func schedule(_ state: State) {
    var scheduleState: ScheduleState = .initial

    let d = self._scheduler.schedule(state) { state -> Disposable in
        // best effort
        if self._group.isDisposed {
            return Disposables.create()
        }
        // 这里因为在递归环境，加了一把锁递归锁，保障安全 
        let action = self._lock.calculateLocked { () -> Action? in
            switch scheduleState {
            case let .added(removeKey):
                self._group.remove(for: removeKey)
            case .initial:
                break
            case .done:
                break
            }

            scheduleState = .done

            return self._action
        }
        
        if let action = action {
            action(state, self.schedule)
        }
        
        return Disposables.create()
    }
    
    ......
}

• 首先为了保证线程安全，保证FIFO，加了一把锁。这也就是为什么RxSwift的信号执行是有顺序的
• 然后执行action,也就是外界传给递归调度者的闭包，后面就是发送信号的常规流程self.forwardOn(.next(next))。
• ps:RxSwift的代码调用非常的繁琐，嵌套很深，各种闭包，所以需要慢慢地一遍一遍的打断点去仔细斟酌

使用 observeOn

• 我们用 observeOn 来决定在哪个 Scheduler 监听这个数据序列。以上例子中，通过使用 observeOn 方法切换到主线程来监听并且处理结果。
• observeOn 操作符将指定一个不同的 Scheduler 来让 Observable 通知观察者。
• observeOn大体流程和思想跟subscribeOn差不多，所以这里就不一一分析了。

总结

• 调度器 Scheduler 的继承关系图：
• Schedulers 是 Rx 实现多线程的核心模块，它主要用于控制任务在哪个线程或队列运行
• subscribeOn 来决定数据序列的构建函数在哪个 Scheduler 上运行
• observeOn 来决定在哪个 Scheduler 监听这个数据序列
• subscribeOn和observeOn都会创建一个中间层序列，所以内部也有一个订阅响应序列的流程，中间层的 sink 就是源序列的观察者

有问题或者建议和意见，欢迎大家评论或者私信。
喜欢的朋友可以点下关注和喜欢，后续会持续更新文章。