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

就问此时此刻还有谁？45度仰望天空，该死！我这无处安放的魅力！

• RxSwift（1）— 初探
• RxSwift（2）— 核心逻辑源码分析
• RxSwift（3）— Observable序列的建立方式
• RxSwift（4）— 高阶函数（上）
• RxSwift（5）— 高阶函数（下）
• RxSwift（6）— 调度者-scheduler源码解析（上）
• RxSwift（7）— 调度者-scheduler源码解析（下）
• RxSwift（8）— KVO底层探索（上）
• RxSwift（9）— KVO底层探索（下）
• RxSwift（10）— 场景序列总结
• RxSwift（11）— dispose源码解析
• RxSwift（12）— Subject即攻也守
• RxSwift（13）— 爬过的坑
• RxSwift（14）— MVVM双向绑定

---

RxSwift 目录直通车 --- 和谐学习，不急不躁！

---

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 实现多线程的核心模块，它主要用于控制任务在哪一个线程或队列运行。 最后附上一张调度者继承关系图，但愿但愿研究的小伙伴继续加油，一路坚持一路花开 一一 和谐学习，不急不躁

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

就问此时此刻还有谁？45度仰望天空，该死！我这无处安放的魅力！