03RxSwift中的timer浅析（二）

注：源码展示只展示关键部分，其余用...代替

RXSwift中的timer不受runloop的影响

首先timer的创建， 常规操作有几种：

swift基本操作，需要加入到common中。

 timer = Timer.init(timeInterval: 1, target: self, selector: #selector(tiemFire), userInfo: nil, repeats: true);
        RunLoop.current.add(timer, forMode: .common);

GCD，GCD的timer是不受runloop的影响的，为什么，去看GCD~~~0.0

gcdTimer = DispatchSource.makeTimerSource()
        gcdTimer?.schedule(deadline: DispatchTime.now(), repeating: DispatchTimeInterval.seconds(1))
        gcdTimer?.setEventHandler(handler: {
            print("GCD Timer")
        })
gcdTimer?.resume()//默认是挂起状态

CADisplayLink，根据屏幕刷新的timer方法

cadTimer = CADisplayLink(target: self, selector: #selector(timerFire))
        cadTimer?.preferredFramesPerSecond = 1
        cadTimer?.add(to: RunLoop.current, forMode: .default)

RxSwift，不受runloop影响和实现原理就是我们今天要研究的问题

var timer:Observable!
timer = Observable<Int>.timer(1, period: 1, scheduler: MainScheduler.instance)//放在主线程执行 MainScheduler.instance
        timer.subscribe(onNext: { (num)in
            print(num)
        })
        .disposed(by:disposeBag)

其实rx的timer内部对GCD进行了封装，查看timer内部实现（代码内部写了注释）：

public static func timer(_ dueTime: RxTimeInterval, period: RxTimeInterval? = nil, scheduler: SchedulerType)
        -> Observable<Element> {
        return Timer(
            dueTime: dueTime,
            period: period,
            scheduler: scheduler
        )
    }

______________进入到Timer_________________
final private class Timer<Element: RxAbstractInteger>: Producer<Element> {
    ...
//初始化
    init(dueTime: RxTimeInterval, period: RxTimeInterval?, scheduler: SchedulerType) {
       ...
    }
//run方法
    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        if self._period != nil {
            let sink = TimerSink(parent: self, observer: observer, cancel: cancel)
            let subscription = sink.run()
            return (sink: sink, subscription: subscription)
        }
        else {
            let sink = TimerOneOffSink(parent: self, observer: observer, cancel: cancel)
            let subscription = sink.run()
            return (sink: sink, subscription: subscription)
        }
    }
}

____Timer继承了Producer，所以最后一定会进入run方法（不清楚的需要去了解rx实现的核心逻辑）_____
 func run() -> Disposable {
        return self._parent._scheduler.schedulePeriodic(0 as Observer.Element, startAfter: self._parent._dueTime, period: self._parent._period!) { state in
            self._lock.lock(); defer { self._lock.unlock() }
            self.forwardOn(.next(state))
            return state &+ 1
            //{}内标记为闭包1
        }
    }

---------------多次跳转，进入方法内部----------------
func schedulePeriodic<StateType>(_ state: StateType, startAfter: RxTimeInterval, period: RxTimeInterval, action: @escaping (StateType) -> StateType) -> Disposable {
        let initial = DispatchTime.now() + startAfter

        var timerState = state
        //1.这里就可以看到，为什么rx的timer是封装的GCD
        let timer = DispatchSource.makeTimerSource(queue: self.queue)
        timer.schedule(deadline: initial, repeating: period, leeway: self.leeway)
        
       
        var timerReference: DispatchSourceTimer? = timer
        let cancelTimer = Disposables.create {
            timerReference?.cancel()
            timerReference = nil
        }
        //2.这个就是我们需要研究的方法，通过action(timerState)不断的发送状态
        timer.setEventHandler(handler: {
            if cancelTimer.isDisposed {
                return
            }
            timerState = action(timerState)
        })
        timer.resume()
        
        return cancelTimer
    }
}

通过以上代码，得出rx的timer是封装的GCD

下面对action(timerState)方法进行探究：
timerState参数就是一个状态，主要是看action方法的实现，action其实是一个逃逸闭包@escaping (StateType) -> StateType)，这个闭包是通过方法一步步传进来的，从方法self._parent._scheduler.schedulePeriodic(0 as Observer.Element, startAfter: self._parent._dueTime, period: self._parent._period!) { state in self._lock.lock(); defer { self._lock.unlock() } self.forwardOn(.next(state)) return state &+ 1 //{}内标记为闭包1 }一直到了schedulePeriodic<StateType>(_ state: StateType, startAfter: RxTimeInterval, period: RxTimeInterval, action: @escaping (StateType) -> StateType) -> Disposable，所以得出结论，这个action，就是我上面所标记的闭包1,就是如下的代码块：

{ state in
            self._lock.lock(); defer { self._lock.unlock() }
            self.forwardOn(.next(state))
            return state &+ 1
            //{}内标记为闭包1
        }

执行self.forwardOn(.next(state))方法，进入到

class Sink<Observer: ObserverType> : Disposable {
   ...
    final func forwardOn(_ event: Event<Observer.Element>) {
        #if DEBUG
            self._synchronizationTracker.register(synchronizationErrorMessage: .default)
            defer { self._synchronizationTracker.unregister() }
        #endif
        if isFlagSet(self._disposed, 1) {
            return
        }
        self._observer.on(event)
    }
...
}

来到了sink的forwardOn，后面再执行 self._observer.on(event)，这个流程又回到了rx的核心逻辑，就不在赘叙。

(大致流程提一嘴：self._observer..on中self._observer是在timer.subscribe方法创建的，去查看subscribe方法中_observer的创建let observer = AnonymousObserver<Element>，on方法在哪里呢？在class AnonymousObserver<Element>: ObserverBase<Element>的父类ObserverBase方法中，on方法中调用了self.onCore(event),这个self.onCore(event)在AnonymousObserver中，方法又调用了self._eventHandler(event)，也就是调用了let observer = AnonymousObserver<Element>后面对应的闭包，然后通过switch方法去选择执行，然后打印输出，完事。)