storm定时器timer源码分析-timer.clj

storm定时器与java.util.Timer定时器比较相似。java.util.Timer定时器实际上是个线程,定时调度所拥有的TimerTasks;storm定时器也有一个线程负责调度所拥有的"定时任务"。storm定时器的"定时任务"是一个vector类型的数据[time, callback, uuid],内有会有三个值,分别是时间、函数、和uuid,很好理解,时间表示该定时任务什么时候执行,函数表示要执行的函数,uuid用于标识该"定时任务"。"定时任务"被存放到定时器的PriorityQueue队列中(和PriorityBlockingQueue区别,在于没有阻塞机制,不是线程安全的)。优先级队列是堆数据结构的典型应用,如果不提供Comparator的话,优先队列中元素默认按自然顺序排列,也就是数字默认是小的在队列头,字符串则按字典序排列(参阅 Comparable),也可以根据 Comparator 来指定,这取决于使用哪种构造方法。优先级队列不允许null元素。依靠自然排序的优先级队列还不允许插入不可比较的对象(这样做可能导致 ClassCastException)。当然也可以自己重新实现Comparator接口, 比如storm定时器就用reify重新实现了Comparator接口。storm定时器的执行过程比较简单,通过timer-thread,不断检查PriorityQueue里面时间最小的"定时任务"是否已经可以触发了, 如果可以(当前时间>=执行时间),就poll出来,调用callback,并sleep。storm定时器相关的函数均定义在timer.clj文件中,storm定时器是由mk-timer函数创建的,mk-timer函数定义如下:

mk-timer函数
()
       ))))
                            (catch Throwable t
                              ;; Because the interrupted exception can be
                              ;; wrapped in a RuntimeException.
                              ;; 检查是否是InterruptedException,如果是InterruptedException,说明线程是由于接收interrupt信号而中断的,不做异常处理,否则调用kill-fn函数、修改线程状                               ;; 态并抛出该异常
                              (when-not (exception-cause? InterruptedException t)
                                (kill-fn t)
                                (reset! active false)
                                (throw t)))))
                        ;; release notifier信号量,标识timer—thread运行结束
                        (.release notifier)) thread-name)]
   ;; 设置timer-thread为守护线程
   (.setDaemon timer-thread true)
   ;; 设置timer-thread为最高优先级
   (.setPriority timer-thread Thread/MAX_PRIORITY)
   ;; 启动timer-thread线程
   (.start timer-thread)
   ;; 返回该定时器的"属性"
   {:timer-thread timer-thread
    :queue queue
    :active active
    :lock lock
    :cancel-notifier notifier}))

我们可以通过调用cancel-timer函数中断一个timer-thread线程,cancel-timer函数定义如下:

cancel-timer函数
(defn cancel-timer
 [timer]
 ;; 检查timer状态是否是"active",如果不是则抛出异常
 (check-active! timer)
 ;; 加锁
 (locking (:lock timer)
   ;; 将timer的状态active设置成false,即"dead"
   (reset! (:active timer) false)
   ;; 调用interrupt方法,中断线程,通过mk-timer函数我们可以知道在线程的run方法内调用了sleep方法,当接收到中断新号后会抛出InterruptedException异常使线程退出
   (.interrupt (:timer-thread timer)))
 ;; acquire timer中的notifier信号量,因为只有当线程结束前才会release notifier信号量,所以此处是等待线程结束
 (.acquire (:cancel-notifier timer)))

check-active!函数定义如下:

check-active!函数
(defn- check-active!
 [timer]
 (when-not @(:active timer)
   (throw (IllegalStateException. "Timer is not active"))))

通过调用schedule函数和schedule-recurring函数我们可以向storm定时器中添加"定时任务"。schedule函数定义如下:

schedule函数
(defnk schedule
 ;; timer绑定定时器,delay-secs绑定"定时任务"相对当前时间的延迟时间,afn绑定callback函数,check-active是否需要检查定时器
 [timer delay-secs afn :check-active true]
 ;; 检查定时器状态
 (when check-active (check-active! timer))
 (let [id (uuid)
       ^PriorityQueue queue (:queue timer)]
   ;; 加锁,执行时间=当前时间+延迟时间,将"定时任务"的vector类型数据添加到PriorityQueue队列中
   (locking (:lock timer)
     (.add queue [(+ (current-time-millis) (secs-to-millis-long delay-secs)) afn id]))))

schedule-recurring函数定义如下:schedule-recurring函数也很简单,与schedule函数的区别就是在"定时任务"的callback函数中又添加了一个相同的"定时任务"。schedule函数的语义可以理解成向定时器添加

一个"一次性任务",schedule-recurring函数的语义可以理解成向定时器添加"一个周期执行的定时任务"(开始执行时间=当前时间+延迟时间,然后每隔recur-secs执行一次),

schedule-recurring函数
(defn schedule-recurring
 [timer delay-secs recur-secs afn]
 (schedule timer
           delay-secs
           (fn this []
             (afn)
             ; This avoids a race condition with cancel-timer.
             (schedule timer recur-secs this :check-active false))))

nimbus检查心跳和重分配任务的实现就是通过schedule-recurring函数向storm定时器添加了一个"周期任务"实现的。

(
       (conf NIMBUS-MONITOR-FREQ-SECS)
       (fn []
         (when (conf NIMBUS-REASSIGN)
           (locking (:submit-lock nimbus)
             (mk-assignments nimbus)))
         (do-cleanup nimbus)
         ))
上一篇:P1001


下一篇:【RL-TCPnet网络教程】第40章 RL-TCPnet之TFTP客户端(精简版)