mk-timer
timer是基于PriorityQueue实现的(和PriorityBlockingQueue区别, 在于没有阻塞机制, 不是线程安全的), 优先级队列是堆数据结构的典型应用
默认情况下, 按照自然顺序(其实就是默认comparator的定义), 最小的元素排在堆头
当然也可以自己重新实现comparator接口, 比如timer就用reify重新实现了comparator接口
整个过程其实比较简单, 开个timer-thread, 不断check PriorityQueue里面时间最小的timer是否已经可以触发
如果可以, 就poll出来, 调用callback, 并sleep, 都很好理解
唯一需要说的是, 这里使用Semaphore,
信号量和lock相似, 都是用于互斥
不同在于, 信号量模拟资源管理, 所以不同于lock的排他, 信号量可以接收多个aquire(取决于配置)
另外一个比较大的区别, lock是解铃还须系铃人, 谁锁谁解, 而信号量无所谓, 任何线程都可以调用release, 或acquire
这里使用信号量, 是用于在cancel-timer时, 等待timer-thread结束
(defn cancel-timer [timer]
(check-active! timer)
(locking (:lock timer)
(reset! (:active timer) false)
(.interrupt (:timer-thread timer)))
(.acquire (:cancel-notifier timer)))
因为cancel的过程就是将active置false, 然后就是调用acquire等待信号量cancel-notifier被释放
而timer-thread在线程结束前, 会release这个信号量
(defnk mk-timer [:kill-fn (fn [& _] )]
(let [queue (PriorityQueue. 10
(reify Comparator
(compare [this o1 o2]
(- (first o1) (first o2))
)
(equals [this obj]
true
)))
active (atom true) ;;标志位
lock (Object.) ;;创建lock对象, 由于PriorityQueue非线程安全, 所以使用locking来保证同时只有一个线程访问queue
notifier (Semaphore. 0) ;;创建信号量, 初始为0
timer-thread (Thread.
(fn []
(while @active
(try
;;peek读但不从queue中取出, 先读出time看看, 符合条件再取出
(let [[time-secs _ _ :as elem] (locking lock (.peek queue))]
(if (and elem (>= (current-time-secs) time-secs))
;;无法保证恰好, 只要当前时间>=time-secs, 就可以执行, 可想而知对于afn必须不能耗时, 否则会影响其他timer
;; imperative to not run the function inside the timer lock
;; otherwise, it's possible to deadlock if function deals with other locks
;; (like the submit lock)
(let [afn (locking lock (second (.poll queue)))] ;;poll从queue中取出
(afn)) ;;真正执行timer中的callback
(Time/sleep 1000)
))
(catch Throwable t
;; because the interrupted exception can be wrapped in a runtimeexception
(when-not (exception-cause? InterruptedException t)
(kill-fn t)
(reset! active false)
(throw t))
)))
(.release notifier)))]
(.setDaemon timer-thread true)
(.setPriority timer-thread Thread/MAX_PRIORITY)
(.start timer-thread)
{:timer-thread timer-thread
:queue queue
:active active
:lock lock
:cancel-notifier notifier}))
schedule
schedule其实就是往PriorityQueue里面插入timer
对于循环schdule, 就是在timer的callback里面, 再次schedule
(defnk schedule [timer delay-secs afn :check-active true]
(when check-active (check-active! timer))
(let [id (uuid)
^PriorityQueue queue (:queue timer)]
(locking (:lock timer)
(.add queue [(+ (current-time-secs) delay-secs) afn id])
)))
(defn schedule-recurring [timer delay-secs recur-secs afn]
(schedule timer
delay-secs
(fn this []
(afn)
(schedule timer recur-secs this :check-active false)) ; this avoids a race condition with cancel-timer
))
使用例子
Supervisor中的使用例子, 定期的调用hb函数更新supervisor的hb
在mk-timer时, 传入的kill-fn callback, 会在timer-thread发生exception的时候被调用
:timer (mk-timer :kill-fn (fn [t]
(log-error t "Error when processing event")
(halt-process! 20 "Error when processing an event")
))
(schedule-recurring (:timer supervisor)
0
(conf SUPERVISOR-HEARTBEAT-FREQUENCY-SECS)
heartbeat-fn)
本文章摘自博客园，原文发布日期：2013-07-02
