RxSwift-Scheduler

Scheduler简介

Scheduler，又称调度者，有一下分类：

• CurrentThreadScheduler - 当前线程
• MainScheduler - 表示主线程，当需要执行一些UI相关任务时，需要切换到该Scheduler下执行
• SerialDispatchQueueScheduler - 封装了GCD的串行队列，需要执行一些串行任务可以切换到这个Scheduler下运行
• ConcurrentDispatchQueueScheduler - 封装了GCD的并行队列，需要执行一些并行任务时，可以切换到这个Scheduler下执行
• OperationQueueScheduler - 封装了NSOperationQueue

源码解析observeOn-ConcurrentDispatchQueueScheduler

使用

let ob = Observable.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
ob.observeOn(ConcurrentDispatchQueueScheduler.init(qos: .background))
    .subscribe{
        print("observeOn-ConcurrentDispatchQueueScheduler:\($0)-",Thread.current)
    }.disposed(by: disposeBag)

打印

observeOn-ConcurrentDispatchQueueScheduler:next(1)- <NSThread: 0x6000022cd9c0>{number = 3, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(2)- <NSThread: 0x6000022cd9c0>{number = 3, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(3)- <NSThread: 0x6000022cd9c0>{number = 3, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(4)- <NSThread: 0x6000022cd9c0>{number = 3, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(5)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(6)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(7)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(8)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(9)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:next(10)- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}
observeOn-ConcurrentDispatchQueueScheduler:completed- <NSThread: 0x6000022ecdc0>{number = 4, name = (null)}

####源码
//ObserveOn.swift
extension ObservableType {
    public func observeOn(_ scheduler: ImmediateSchedulerType)
        -> Observable<Element> {
            if let scheduler = scheduler as? SerialDispatchQueueScheduler {
                return ObserveOnSerialDispatchQueue(source: self.asObservable(), scheduler: scheduler)
            }
            else {
                return ObserveOn(source: self.asObservable(), scheduler: scheduler)
            }
    }
}

创建ObserveOn内部类

final private class ObserveOn<Element>: Producer<Element> {
    let scheduler: ImmediateSchedulerType
    let source: Observable<Element>

    init(source: Observable<Element>, scheduler: ImmediateSchedulerType) {
        self.scheduler = scheduler
        self.source = source

#if TRACE_RESOURCES
        _ = Resources.incrementTotal()
#endif
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        let sink = ObserveOnSink(scheduler: self.scheduler, observer: observer, cancel: cancel)
        let subscription = self.source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }

#if TRACE_RESOURCES
    deinit {
        _ = Resources.decrementTotal()
    }
#endif
}

ObserveOn这个类继承Producer，所以当序列被订阅时必定执行其run方法

final private class ObserveOnSink<Observer: ObserverType>: ObserverBase<Observer.Element> {
    typealias Element = Observer.Element 

    let _scheduler: ImmediateSchedulerType

    var _lock = SpinLock()
    let _observer: Observer

    // state
    var _state = ObserveOnState.stopped
    var _queue = Queue<Event<Element>>(capacity: 10)

    let _scheduleDisposable = SerialDisposable()
    let _cancel: Cancelable

    init(scheduler: ImmediateSchedulerType, observer: Observer, cancel: Cancelable) {
        self._scheduler = scheduler
        self._observer = observer
        self._cancel = cancel
    }

    override func onCore(_ event: Event<Element>) {
        let shouldStart = self._lock.calculateLocked { () -> Bool in
            self._queue.enqueue(event)

            switch self._state {
            case .stopped:
                self._state = .running
                return true
            case .running:
                return false
            }
        }

        if shouldStart {
            self._scheduleDisposable.disposable = self._scheduler.scheduleRecursive((), action: self.run)
        }
    }

    func run(_ state: (), _ recurse: (()) -> Void) {
        let (nextEvent, observer) = self._lock.calculateLocked { () -> (Event<Element>?, Observer) in
            if !self._queue.isEmpty {
                return (self._queue.dequeue(), self._observer)
            }
            else {
                self._state = .stopped
                return (nil, self._observer)
            }
        }

        if let nextEvent = nextEvent, !self._cancel.isDisposed {
            observer.on(nextEvent)
            if nextEvent.isStopEvent {
                self.dispose()
            }
        }
        else {
            return
        }

        let shouldContinue = self._shouldContinue_synchronized()

        if shouldContinue {
            recurse(())
        }
    }

    func _shouldContinue_synchronized() -> Bool {
        self._lock.lock(); defer { self._lock.unlock() } // {
            if !self._queue.isEmpty {
                return true
            }
            else {
                self._state = .stopped
                return false
            }
        // }
    }

    override func dispose() {
        super.dispose()

        self._cancel.dispose()
        self._scheduleDisposable.dispose()
    }
}

extension ImmediateSchedulerType {
    /**
    Schedules an action to be executed recursively.
    
    - parameter state: State passed to the action to be executed.
    - parameter action: Action to execute recursively. The last parameter passed to the action is used to trigger recursive scheduling of the action, passing in recursive invocation state.
    - returns: The disposable object used to cancel the scheduled action (best effort).
    */
    public func scheduleRecursive<State>(_ state: State, action: @escaping (_ state: State, _ recurse: (State) -> Void) -> Void) -> Disposable {
        let recursiveScheduler = RecursiveImmediateScheduler(action: action, scheduler: self)
        
        recursiveScheduler.schedule(state)
        
        return Disposables.create(with: recursiveScheduler.dispose)
    }
}

思维导图，这个分析到一半感觉有点不对劲，后续再研究

observeOn