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度仰望天空，該死！我這無處安放的魅力！