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

RxSwift 探索了几天，今天晚上会迎来同学们会疑虑比较多，平时比较好奇的模块 - RxSwift调度者 - scheduler. 这个模块相对前面的流程比较复杂，这一篇博客希望每一个同学一定耐着性子跟着我的思维锊清楚。后面总结你会发现原来也就那么一回事！在 RxSwift 的世界如果看得简单一点：那么只有四个东西：

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

下面开始具体分析这个非常有意思的东西 调度者 - Scheduler

调度者的作用 - 封装了一套多线程方案

大家可以想象，多线程对我们 iOS开发 的性能影响是非常大，而对于 RxSwift 这么一套爱装逼的生态，不对线程下手，我觉得不符合它的风格！果然 RxSwift 的大神主要针对 GCD 进行了一套 scheduler 封装。

• CurrentThreadScheduler：表示当前线程 Scheduler。（默认使用这个）

public class CurrentThreadScheduler : ImmediateSchedulerType {
    typealias ScheduleQueue = RxMutableBox<Queue<ScheduledItemType>>
    /// The singleton instance of the current thread scheduler.
    public static let instance = CurrentThreadScheduler()

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

    public func schedule<StateType>(_ state: StateType, action: @escaping (StateType) -> Disposable) -> Disposable {
     // 篇幅原因只贴出重点    
     }
}

• 这个作为代表详细分析，后面相同的内容大家自己类比
• 外界获取判断当前队列的是否被关联：isScheduleRequired
• 利用对 queue的set,get方法的观察，绑定我们的当前队列与静态字符串(这种用法，也是开发比较常用的)
• 下面就是对线程的拓展

extension Thread {
    static func setThreadLocalStorageValue<T: AnyObject>(_ value: T?, forKey key: NSCopying) {
        let currentThread = Thread.current
        let threadDictionary = currentThread.threadDictionary

        if let newValue = value {
            threadDictionary[key] = newValue
        }
        else {
            threadDictionary[key] = nil
        }
    }

    static func getThreadLocalStorageValueForKey<T>(_ key: NSCopying) -> T? {
        let currentThread = Thread.current
        let threadDictionary = currentThread.threadDictionary
        
        return threadDictionary[key] as? T
    }
}

• MainScheduler：表示主线程。如果我们需要执行一些和 UI 相关的任务，就需要切换到该 Scheduler 运行。一看下面的源码就非常清晰，绑定了主队列DispatchQueue.main

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

• 同是这里还有继承了SerialDispatchQueueScheduler就是串行调度者，其实我们也是可以理解的，主队列其实就是一种串行队列。

public class SerialDispatchQueueScheduler : SchedulerType {
    let configuration: DispatchQueueConfiguration
    init(serialQueue: DispatchQueue, leeway:) {
        self.configuration = DispatchQueueConfiguration(queue: leeway:)
    }

public convenience init(internalSerialQueueName: serialQueueConfiguration: leeway: ) {
        let queue = DispatchQueue(label: internalSerialQueueName, attributes: [])
        serialQueueConfiguration?(queue)
        self.init(serialQueue: queue, leeway: leeway)
    }
 }

• 从这里也可以看出就是接受串行队列
• 如果没有，自己内部创建一个串行队列
• ConcurrentDispatchQueueScheduler的思路和SerialDispatchQueueScheduler 也是一模模一样样

public class ConcurrentDispatchQueueScheduler: SchedulerType {
    public typealias TimeInterval = Foundation.TimeInterval
    public typealias Time = Date
    
    public var now : Date {
        return Date()
    }

    let configuration: DispatchQueueConfiguration
  
    public init(queue: leeway: ) {
        self.configuration = DispatchQueueConfiguration(queue: leeway:)
    }
    
    public convenience init(qos: leeway: ) {
        self.init(queue: DispatchQueue(
            label: "rxswift.queue.\(qos)",
            qos: qos,
            attributes: [DispatchQueue.Attributes.concurrent],
            target: nil),
            leeway: leeway
        )
    }
}

• OperationQueueScheduler：封装了 NSOperationQueue, 下面代码非常清晰了，典型的操作队列和操作优先级

public class OperationQueueScheduler: ImmediateSchedulerType {
    public let operationQueue: OperationQueue
    public let queuePriority: Operation.QueuePriority
    public init(operationQueue: queuePriority: ) {
        self.operationQueue = operationQueue
        self.queuePriority = queuePriority
    }
}

调度执行

上面我们已经大致了解了线程，队列的封装(就是换了一个马甲)，其实如果你是一个用心的开发人员，看到这里，你肯定非常关系。到底我们的调度执行是怎么样的呢？下面我们直接分析，可以说简单的让你发指，但是真正难起来，又是让你发脱~~~~~哈哈哈哈哈

我们每次初始化是不是都有个非常重要的东西就是:
let configuration: DispatchQueueConfiguration 这里保存了我们需要的队列以及leeway信息，一般我们开发能保存信息的家伙都不简单

func schedule<StateType>(_ state: action: ) -> Disposable {
    return self.scheduleInternal(state, action: action)
}

func scheduleInternal<StateType>(_ state:  action: ) -> Disposable {
    return self.configuration.schedule(state, action: action)
}

func scheduleRelative<StateType>(_ state: dueTime: action: ) -> Disposable {
    return self.configuration.scheduleRelative(state, dueTime: action:)
}

func schedulePeriodic<StateType>(state: startAfter:period: action: ) -> Disposable {
    return self.configuration.schedulePeriodic(state, startAfter: period: action:)
}

• 从上面核心方法：schedule 可以非常轻松看出都是我们的 self.configuration具体施行者，下面我们来分析内部流程！

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

• 我的天啊！还要怎么明显！就是在当前队列下面，异步调度执行具体的：action(state)

分析完了，哈哈哈！但是如果你是一个讲究的人，一定会有这样的疑问，什么时候调用 scheduler 上下游流程又是怎么样的？这是目前我们不得而知的。因为还有一篇恐怖的未知领域在等着你！Ready? 请进入下一篇：RxSwift调度者-scheduler源码解析（下）

总结

调度器（Schedulers）是 RxSwift 实现多线程的核心模块，它主要用于控制任务在哪个线程或队列运行。
最后附上一张调度者继承关系图，希望希望研究的小伙伴继续加油，一路坚持一路花开 一一 和谐学习，不急不躁