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

;; kill-fn函数会在timer-thread发生exception的时候被调用，timer-name标识定时器的名称
( defnk mk-timer [ :kill-fn ( fn [ & _ ] ) :timer-name nil ]
  ;; queue绑定PriorityQueue队列，创建PriorityQueue队列时指定队列初始容量为10，并指定一个Comparator比较器，Comparator比较器比较"定时任务"执行时间的大小，这样每次poll出执行时间最小的"定  时任务"， PriorityQueue队列是一个依赖执行时间的小顶堆
 ( let [ queue ( PriorityQueue. 10 ( reify Comparator
                                  ( compare
                                    [ this o1 o2 ]
                                    ( - ( first o1) ( first o2)))
                                  ( equals
                                    [ this obj ]
                                    true)))
        ;; active标识timer-thread是"active"的
        active ( atom true)
        ;; 创建一个锁，因为PriorityQueue并不是线程安全的，所以通过这个锁，可以使多线程互斥访问PriorityQueue
        lock ( Object.)
        ;; notifier是一个java信号量，初始值为0，notifier信号量的主要功能就是当我们调用cancel-timer函数中断一个timer-thread时，等待timer-thread结束，当timer-thread结束前会release          ;; notifier信号量
        notifier ( Semaphore. 0)
        ;; thread-name绑定timer-thread线程名，没有指定时默认为"timer"
        thread-name ( if timer-name timer-name "timer")
        ;; timer-thread线程
        timer-thread ( Thread.
                      ( fn []
                        ;; 当timer-thread为"active"即active=true时，进入while循环
                        ( while @ active
                          ( try
                            ;; peek函数从PriorityQueue获取执行时间最小的"定时任务"，但并不出队列。time-millis绑定执行时间，elem绑定"定时任务"数据
                            ( let [[ time-millis _ _ :as elem ] ( locking lock ( .peek queue ))]
                              ;; 如果elem不为nil且当前时间>=执行时间，那么先加锁，然后poll出该"定时任务"，并将"定时任务"的callback函数绑定到afn，最后调用该函数；否则判断time-millis                               ;; 是否为nil
                              ;; 我们可以发现该定时器是软时间执行"定时任务"的，也就是说"定时任务"有可能被延迟执行，同时如果afn函数执行时间比较长，那么会影响下一个"定时任务"的执行
                              ( if ( and elem ( >= ( current-time-millis) time-millis))
                                ;; It is imperative to not run the function
                                ;; inside the timer lock. Otherwise, it is
                                ;; possible to deadlock if the fn deals with
                                ;; other locks, like the submit lock.
                                ( let [ afn ( locking lock ( second ( .poll queue )))]
                                  ;; 执行"定时任务"的callback函数
                                  ( afn))
                                ;; 该if语句是上面if语句的else分支，判断time-millis是否为nil，如果time-millis不为nil，则timer-thread线程sleep(执行时间-当前时间)；否则                                             ;; sleep(1000)，表明PriorityQueue中没有"定时任务"
                                ( if time-millis
                                  ;; If any events are scheduled, sleep until
                                  ;; event generation. If any recurring events
                                  ;; are scheduled then we will always go
                                  ;; through this branch, sleeping only the
                                  ;; exact necessary amount of time.
                                  ( Time/sleep ( - time-millis ( current-time-millis)))
                                  ;; Otherwise poll to see if any new event
                                  ;; was scheduled. This is, in essence, the
                                  ;; response time for detecting any new event
                                  ;; schedulings when there are no scheduled
                                  ;; events.
                                  ( Time/sleep 1000))))
                            ( 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 }))

( 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)))

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

( 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 ]))))

( 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))))

( schedule-recurring ( :timer nimbus)
        0
       ( conf NIMBUS-MONITOR-FREQ-SECS)
       ( fn []
         ( when ( conf NIMBUS-REASSIGN)
           ( locking ( :submit-lock nimbus)
             ( mk-assignments nimbus)))
         ( do-cleanup nimbus)
         ))