Storm-源码分析-EventManager (backtype.storm.event)

Protocol and DataType

大体结构,

定义protocol EventManager, 其实就是定义interface

函数event-manager, 主要做2件事
1. 启动event queue的处理线程, 不断从queue中取出event-fn并执行
2. 返回实现EventManager的匿名record(reify部分, 实现protocol)

这里使用了reify的close over特性, reify会将用到的局部变量打包到闭包内, 包含queue, runner

(ns backtype.storm.event
  (:use [backtype.storm log util])
  (:import [backtype.storm.utils Time Utils])
  (:import [java.util.concurrent LinkedBlockingQueue TimeUnit])
  )

(defprotocol EventManager
  (add [this event-fn])
  (waiting? [this])
  (shutdown [this]))

(defn event-manager
  "Creates a thread to respond to events. Any error will cause process to halt"
  [daemon?]
  (let [added (atom 0)
        processed (atom 0)
        ^LinkedBlockingQueue queue (LinkedBlockingQueue.)
        running (atom true)
        runner (Thread.
                  (fn []
                    (try-cause
                      (while @running
                        (let [r (.take queue)]
                          (r)
                          (swap! processed inc)))
                    (catch InterruptedException t
                      (log-message "Event manager interrupted"))
                    (catch Throwable t
                      (log-error t "Error when processing event")
                      (halt-process! 20 "Error when processing an event"))
                      )))]
    (.setDaemon runner daemon?)
    (.start runner)
    (reify
      EventManager
      (add [this event-fn]
        ;; should keep track of total added and processed to know if this is finished yet
        (when-not @running
          (throw (RuntimeException. "Cannot add events to a shutdown event manager")))
        (swap! added inc)
        (.put queue event-fn)
        )
      (waiting? [this]
        (or (Time/isThreadWaiting runner)
            (= @processed @added)
            ))
      (shutdown [this]
        (reset! running false)
        (.interrupt runner)
        (.join runner)
        )
        )))

 

使用的时候很简单, 如下

let [event-manager processes-event-manager :as managers] [(event/event-manager false) (event/event-manager false)]
(.add processes-event-manager sync-processes)

可以直接调用add或其他的function
相当于给event-manager增加EventManager protocol, 反过来说, 给add或其他接口functions增加对event-manager record的support, 因为protocol函数的第一个参数总是类型

比较神奇的是, 闭包产生的效果, 可以在完全没有queue, runner定义或声明的情况下, 方便的操作他们