RxSwift 源碼解析04： Timer 源碼解析

本文主要探究 RxSwift 中 timer 的源碼

首先創(chuàng)建一個(gè)timer，timer的創(chuàng)建方式有兩種

  • 自定義timer，即通過(guò) timer函數(shù)創(chuàng)建
Observable<Int>.timer(.seconds(5), period: .seconds(2), scheduler: MainScheduler.instance)
    .subscribe { event in
        print(event)
    }
    .disposed(by: disposeBag)
  • 使用 RxSwift 中封裝好的timer，即通過(guò) interval 創(chuàng)建
var timer: Observable<Int> = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance)
timer.subscribe { num in
    print(num)
}
.disposed(by: disposeBag)

下面我們來(lái)進(jìn)行一一探索

timer() 源碼解析

  • 進(jìn)入函數(shù) timer() 源碼，創(chuàng)建一個(gè) Timer 類對(duì)象（記為 A），有三個(gè)屬性 scheduler（調(diào)度環(huán)境）、dueTime（延遲時(shí)間）、period（時(shí)間間隔）。
extension ObservableType where Element: RxAbstractInteger {

    public static func timer(_ dueTime: RxTimeInterval, period: RxTimeInterval? = nil, scheduler: SchedulerType)
        -> Observable<Element> {
        return Timer(
            dueTime: dueTime,
            period: period,
            scheduler: scheduler
        )
    }
}
  • 查看 Timer 類的繼承鏈：Timer --> Producer --> Observable --> 遵循 ObservableType 協(xié)議 --> 遵循 ObservableConvertibleType 協(xié)議。所有 Timer 類其本質(zhì)也是一個(gè)可觀察序列。
  • 進(jìn)入函數(shù) subscribe() 源碼，創(chuàng)建 AnonymousObserver 對(duì)象（記為 B），并保存訂閱信號(hào)的閉包 eventHandler
extension ObservableType {
    ......

    public func subscribe(
        onNext: ((Element) -> Void)? = nil, // element是Observablecreate創(chuàng)建時(shí)外部傳進(jìn)來(lái)的
        onError: ((Swift.Error) -> Void)? = nil,
        onCompleted: (() -> Void)? = nil,
        onDisposed: (() -> Void)? = nil
    ) -> Disposable {
            ...
            //創(chuàng)建匿名觀察者
            let observer = AnonymousObserver<Element> { event in
                
                ...
            }
            //創(chuàng)建銷(xiāo)毀者
            return Disposables.create(
                // self.asObservable() 是一個(gè)序列(即統(tǒng)一成序列)，直接返回self，即ob
                self.asObservable().subscribe(observer),
                disposable
            )
    }
}
  • 進(jìn)入 AnonymousObservable 類的 subscribe 函數(shù)，最終在父類 Producer 中找到，即 A.subscribe(B) 
class Producer<Element>: Observable<Element> {
    ......

    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        // 調(diào)度者
        if !CurrentThreadScheduler.isScheduleRequired {
            // The returned disposable needs to release all references once it was disposed.
            let disposer = SinkDisposer() //銷(xiāo)毀者
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
                disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

                return disposer
            }
        }
    }
    
    ......
}
  • 進(jìn)入函數(shù) run 的實(shí)現(xiàn)，這個(gè)在 Producer 子類 Timer 類中，源碼如下，即 A.run(B)
    • 這里的 TimerSink 是timer的通道類，用于鏈接 timer 序列 和 observer 觀察者
    • sink在初始化時(shí)，就將 timer 序列作為了 TimerSInk的 parent屬性傳入了，同時(shí)持有 observer（觀察者）、cancel（銷(xiāo)毀者）
final private class Timer<Element: RxAbstractInteger>: Producer<Element> {
    fileprivate let scheduler: SchedulerType //調(diào)度環(huán)境
    fileprivate let dueTime: RxTimeInterval //延遲時(shí)間
    fileprivate let period: RxTimeInterval? //時(shí)間間隔

    init(dueTime: RxTimeInterval, period: RxTimeInterval?, scheduler: SchedulerType) {
        self.scheduler = scheduler
        self.dueTime = dueTime
        self.period = period
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        if self.period != nil {
            //傳入的 self 是 timer
            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)
        }
    }
}
  • 進(jìn)入 TimerSink 類的函數(shù) run() 實(shí)現(xiàn)，這里與 AnonymousObservable 序列的區(qū)別就是少了一個(gè)self傳入，但其實(shí)已經(jīng)在初始化時(shí)將self 傳入了。即 TimerSink.run()，
final private class TimerSink<Observer: ObserverType> : Sink<Observer> where Observer.Element : RxAbstractInteger  {
    typealias Parent = Timer<Observer.Element>

    private let parent: Parent  //timer序列對(duì)象
    private let lock = RecursiveLock() //遞歸鎖

    init(parent: Parent, observer: Observer, cancel: Cancelable) {
        self.parent = parent //保存parent，即Timer
        super.init(observer: observer, cancel: cancel)
    }

    func run() -> Disposable {
        //action 就是 ???? 的閉包
        return self.parent.scheduler.schedulePeriodic(0 as Observer.Element, startAfter: self.parent.dueTime, period: self.parent.period!) { state in
            self.lock.performLocked {
                self.forwardOn(.next(state))
                return state &+ 1 // +1 操作
            }
        }
    }
}
  • 進(jìn)入函數(shù) scheduleRelative 源碼，其中 scheduler 是定時(shí)器初始化時(shí)傳入的定時(shí)器，遵循 SchedulerType 協(xié)議（注：需要根據(jù)具體傳入的線程查找 scheduleRelative 具體實(shí)現(xiàn)，這里我們傳入的是 MainScheduler，所以查找 MainScheduler 的函數(shù) scheduleRelative ），即 SerialDispatchQueueScheduler.scheduleRelative()
    • MainScheduler的繼承鏈：MainScheduler --> SerialDispatchQueueScheduler --> 遵循 SchedulerType 協(xié)議
    • 所以根據(jù)繼承鏈，最終在 SerialDispatchQueueScheduler 類中找到 scheduleRelative 的具體實(shí)現(xiàn)
//SerialDispatchQueueScheduler 類
public class SerialDispatchQueueScheduler : SchedulerType {
    ...

    public final func scheduleRelative<StateType>(_ state: StateType, dueTime: RxTimeInterval, action: @escaping (StateType) -> Disposable) -> Disposable {
        self.configuration.scheduleRelative(state, dueTime: dueTime, action: action)
    }

    ...

}
  • 進(jìn)入函數(shù) scheduleRelative() 的實(shí)現(xiàn)，在這里我們可以看到，其本質(zhì)是通過(guò)GCD定義一個(gè)timer，即 DispatchQueueConfiguration.scheduleRelative()
    • 這里的 action 就是 TimerSink 類中run() 函數(shù)中 schedulePeriodic 函數(shù)的閉包，在這里做了 +1 操作
    • 在每次執(zhí)行完定時(shí)器后，走到回調(diào)方法，不斷地發(fā)送 onNext 信號(hào)，并做 +1 操作，以此來(lái)達(dá)到循環(huán)的操作
extension DispatchQueueConfiguration {
    ...

    func scheduleRelative<StateType>(_ state: StateType, dueTime: RxTimeInterval, action: @escaping (StateType) -> Disposable) -> Disposable {
        let deadline = DispatchTime.now() + dueTime

        let compositeDisposable = CompositeDisposable()
        //初始化一個(gè) gcd timer，并指定線程
        let timer = DispatchSource.makeTimerSource(queue: self.queue)
        //設(shè)置執(zhí)行次數(shù)
        /**  
        wallDeadline: 什么時(shí)候開(kāi)始  
        leeway: 調(diào)用頻率,即多久調(diào)用一次  
        */
        timer.schedule(deadline: deadline, leeway: self.leeway)

        // TODO:
        // This looks horrible, and yes, it is.
        // It looks like Apple has made a conceptual change here, and I'm unsure why.
        // Need more info on this.
        // It looks like just setting timer to fire and not holding a reference to it
        // until deadline causes timer cancellation.
        var timerReference: DispatchSourceTimer? = timer
        let cancelTimer = Disposables.create {
            timerReference?.cancel()
            timerReference = nil
        }
        //設(shè)置執(zhí)行回調(diào)
        timer.setEventHandler(handler: {
            if compositeDisposable.isDisposed {
                return
            }
            _ = compositeDisposable.insert(action(state))
            cancelTimer.dispose()
        })
        //執(zhí)行timer
        timer.resume()

        _ = compositeDisposable.insert(cancelTimer)

        return compositeDisposable
    }

    ...
}
  • 進(jìn)入 action 閉包內(nèi)的函數(shù) forwardOn 實(shí)現(xiàn)，這里在父類 Sink 中找到，即 Sink.forwardOn()
    • 這里的self.observer 就是前面?zhèn)魅?Sink observer 觀察者，即 AnonymousObserver 類的對(duì)象
    • 內(nèi)部實(shí)現(xiàn)執(zhí)行了 觀察者的函數(shù) on() 
final private class TimerSink<Observer: ObserverType> : Sink<Observer> where Observer.Element : RxAbstractInteger  {
    ...

    func run() -> Disposable {
        //action 就是 ???? 的閉包
        return self.parent.scheduler.schedulePeriodic(0 as Observer.Element, startAfter: self.parent.dueTime, period: self.parent.period!) { state in
            self.lock.performLocked {
                self.forwardOn(.next(state))
                return state &+ 1 // +1 操作
            }
        }
    }

    ...
}


??

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 為 Sink
        self.observer.on(event)
    }

    ...
}
  • 進(jìn)入 AnonymousObserver 的函數(shù) on() 實(shí)現(xiàn)，在其父類 ObserverBase 中找到，即 ObserverBase.on()
    • 其內(nèi)部執(zhí)行了函數(shù) onCore() 
class ObserverBase<Element> : Disposable, ObserverType {
    ...

    //不斷發(fā)送響應(yīng)的功能
    func on(_ event: Event<Element>) {
        switch event {
        case .next:
            if load(self.isStopped) == 0 {
                self.onCore(event)
            }
        case .error, .completed:
            if fetchOr(self.isStopped, 1) == 0 {
                self.onCore(event)
            }
        }
    }

    ... 
}
  • 進(jìn)入 AnonymousObserver 類的函數(shù) onCore() 的實(shí)現(xiàn)，因?yàn)楦割悰](méi)有具體實(shí)現(xiàn)，所以在子類查找，即 B.eventHandler(event)，會(huì)走到訂閱信號(hào)的事件處理回調(diào)中
final class AnonymousObserver<Element>: ObserverBase<Element> {
    ...

    override func onCore(_ event: Event<Element>) {
        self.eventHandler(event)
    }
    
    ....
}

interval() 源碼解析

  • 進(jìn)入函數(shù) interval 源碼，也是創(chuàng)建了一個(gè) Timer 類對(duì)象，同樣的有三個(gè)屬性
extension ObservableType where Element: RxAbstractInteger {
 
    public static func interval(_ period: RxTimeInterval, scheduler: SchedulerType)
        -> Observable<Element> {
        return Timer(
            dueTime: period,
            period: period,
            scheduler: scheduler
        )
    }
}
  • 后續(xù)分析同 timer一致，這里就不贅述了。

總結(jié)

繼承鏈

  • Timer 類的繼承鏈：Timer --> Producer --> Observable --> 遵循 ObservableType 協(xié)議 --> 遵循 ObservableConvertibleType 協(xié)議。所有 Timer 類其本質(zhì)也是一個(gè)可觀察序列。
  • MainScheduler的繼承鏈：MainScheduler --> SerialDispatchQueueScheduler --> 遵循 SchedulerType 協(xié)議

timer()/interval() 核心邏輯

  • 創(chuàng)建序列 timer()/interval()：創(chuàng)建 Timer 對(duì)象（A）
  • 訂閱信號(hào) subscribe()：創(chuàng)建 AnonymousObserver 對(duì)象（B），保存eventHandler
    • A.subscribe(B)
    • A.run(B)
    • TimerSink.run() ：執(zhí)行通過(guò)方法
    • SerialDispatchQueueScheduler.scheduleRelative()：閉包中作+1操作
    • DispatchQueueConfiguration.scheduleRelative()：初始化gcd定時(shí)器
    • Sink.forwardOn()：發(fā)送信號(hào)
    • ObserverBase.on() ：轉(zhuǎn)發(fā)信號(hào)
    • B.eventHandler(event) ：處理信號(hào)
timer 核心邏輯
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容