Timer
在RxSwift 中Timer 为一个Observable(序列), 我们可以通过设置和订阅这个序列实现计时器的效果.
Timer 序列分为两种类型.
-
一次性执行.
创建一个 Observable 在一段延时后，产生唯一的一个信号.一般用于某些只需要执行一次的延迟操作.如超时判断.
/// 创建重复多次执行序列
/// 参数1: 几秒后开始第一次执行
/// 参数2: 调度者
let timer = Observable<Int>.timer(1, scheduler: MainScheduler.instance)
/// 添加订阅
timer.subscribe { (event) in
// 执行代码
}.disposed(by: disposeBag)
-
重复多次执行.
创建一个 Observable 在一段延时后，每隔一段时间产生一个信号.
/// 创建重复多次执行序列
/// 参数1: 几秒后开始第一次执行
/// 参数2: 重复执行间隔
/// 参数3: 调度者
let timer = Observable<Int>.timer(1, period: 2, scheduler: MainScheduler.instance)
/// 添加订阅
timer.subscribe { (event) in
// 执行代码
}.disposed(by: disposeBag)
Timer 的实现逻辑
在RxSwift中,任何业务都可以分为三部分. 创建序列, 订阅序列, 发送信号.根据这个我们来看一下Timer 的具体实现方式.
1. 创建序列
let timer = Observable<Int>.timer(1, scheduler: MainScheduler.instance)
进入timer 方法中查看实现,可以看到在timer 方法中创建并返回一个Timer对象.
public static func timer(_ dueTime: RxTimeInterval, period: RxTimeInterval? = nil, scheduler: SchedulerType)
-> Observable<E> {
return Timer(
dueTime: dueTime,
period: period,
scheduler: scheduler
)
}
点开Timer 对象,发现Timer 继承着Producer(生产者),在RxSwift 中,序列大部分都是继承者Producer的.我们在设计一些代码时,通过会让功能相似的类继承着同一基类BaseClass,这里可以理解为Producer就是这些序列的基类,在基类中实现了序列的一些基本操作.例如订阅功能Subscribe().
final private class Timer<E: RxAbstractInteger>: Producer<E> {
fileprivate let _scheduler: SchedulerType
fileprivate let _dueTime: RxTimeInterval
fileprivate let _period: RxTimeInterval?
init(dueTime: RxTimeInterval, period: RxTimeInterval?, scheduler: SchedulerType) {
self._scheduler = scheduler
self._dueTime = dueTime
self._period = period
}
}
2. 添加订阅者
/// 添加订阅
timer.subscribe { (event) in
// 执行代码
}.disposed(by: disposeBag)
进入具体实现方法,可以看到在方法中创建了一个匿名订阅者AnonymousObserver,并通过基类Producer 提供的subscribe(observer) 添加到Timer 的序列中.
注: asObservable() 这个方法会返回self. 原因后面再讲.
public func subscribe(_ on: @escaping (Event<E>) -> Void)
-> Disposable {
let observer = AnonymousObserver { e in
on(e)
}
return self.asObservable().subscribe(observer)
}
进行Producer 类查看subscribe 的实现,在方法中调用self.run(observer, cancel: disposer) 这里的run方法由于子类 Timer 已经实现了.所以实际调用的是子类的run方法.
class Producer<Element> : Observable<Element> {
override func subscribe<O : ObserverType>(_ observer: O) -> Disposable where O.E == Element {
if !CurrentThreadScheduler.isScheduleRequired {
// The returned disposable needs to release all references once it was disposed.
let disposer = SinkDisposer()
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
}
}
}
}
在Timer 的 run方法中,我们根据间隔时间self._period 是否为nil 判断它是一次性执行,还是重复多次执行.然后创建对应的Sink对象TimerSink 和 TimerOneOffSink.Sink对象可以理解为结点或者管道,它连接着序列,观察者,调度者,销毁者.在RxSwift 中非常重要.在创建完Sink 对象之后通过Sink.run() 实现最终的连接.
final private class Timer<E: RxAbstractInteger>: Producer<E> {
override func run<O: ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == E {
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)
}
}
}
这里TimerSink(多次执行)和TimerOneOffSink(单次执行)实现逻辑相似,我们以TimerSink为例.进入TimerSink.run()方法中.在该方法中主要做了两件事.
- 创建
Action block任务,在这个任务中,会调用self.forwardOn(.next(state))在forwardOn中会调用创建Sink对象时保存的订阅者obeserver的on方法.在这个on中会执行我们给序列添加订阅者时保存的信号处理block - 将action block和其他参数
state(计数),startAfter(开始时间),period(执行间隔),传入到调度者中.接下来进入发送信号阶段.
final private class TimerSink<O: ObserverType> : Sink<O> where O.E : RxAbstractInteger {
func run() -> Disposable {
let action = { state in
self._lock.lock(); defer { self._lock.unlock() }
self.forwardOn(.next(state))
return state &+ 1
}
return self._parent._scheduler.schedulePeriodic(0 as O.E, startAfter: self._parent._dueTime, period: self._parent._period!, action: action)
}
}
3. 发送信号
我们进入到调度者中的方法具体实现:可以看到在这个方法中.通过创建一个GDC的Timer,并给这个Timer设置对应参数开始时间,间隔,允许误差和执行任务timer.setEventHandler.在setEventHandler中我们调用了传进入了Action Block任务,在这个任务中会调用forwardOn给订阅者发送信号.到此,整个Timer 的序列创建,订阅,发送信号就解析完毕了!!!
func schedulePeriodic<StateType>(_ state: StateType, startAfter: TimeInterval, period: TimeInterval, action: @escaping (StateType) -> StateType) -> Disposable {
let initial = DispatchTime.now() + dispatchInterval(startAfter)
var timerState = state
let timer = DispatchSource.makeTimerSource(queue: self.queue)
timer.schedule(deadline: initial, repeating: dispatchInterval(period), leeway: self.leeway)
// TODO:
// This looks horrible, and yes, it is.
// It looks like Apple has made a conceputal 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
}
timer.setEventHandler(handler: {
if cancelTimer.isDisposed {
return
}
timerState = action(timerState)
})
timer.resume()
return cancelTimer
}
Timer 思维导图
