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

末蓝、 2024-04-17 13:59 42阅读 0赞

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

> 对于RxSwift，有更好的见解，想要更好的探讨，可以进入iOS技术群，一起探讨交流

### [点击此处进交流群 有技术的来闲聊 没技术的来学习][Link 1] ###

*  可观察序列 - `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源码解析（下）][RxSwift_-scheduler]

## 总结 ##

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

### 小编这里有大量的书籍和面试资料哦（[点击下载][Link 1]） ###

原文链接：[www.jianshu.com/p/6a63f7353…][www.jianshu.com_p_6a63f7353]

转载于:https://juejin.im/post/5d4277e0518825673a6adbdc

[Link 1]: https://link.juejin.im?target=https%3A%2F%2Fjq.qq.com%2F%3F_wv%3D1027%26amp%3Bk%3D5kFIHOq
[RxSwift_-scheduler]: https://link.juejin.im?target=https%3A%2F%2Fwww.jianshu.com%2Fp%2F6157d92db70e
[www.jianshu.com_p_6a63f7353]: https://link.juejin.im?target=https%3A%2F%2Fwww.jianshu.com%2Fp%2F6a63f7353ad4