storm事件管理器定义在event.clj中，主要功能就是通过独立线程执行"事件处理函数"。我们可以将"事件处理函数"添加到EventManager的阻塞队列中，EventManager的事件处理线程不断地从阻塞队列中获取"事件处理函数"并执行。
EventManager协议
协议就是一组函数定义的集合，协议中函数的第一个参数必须为实现该协议的实例本身，类似于java中实例方法的第一个参数为this；协议类似于java中的接口。
(defprotocol EventManager
(add [this event-fn])
(waiting? [this])
(shutdown [this]))
(add [this event-fn])
(waiting? [this])
(shutdown [this]))
event-manager函数
()
)
"Error when processing an event")))))]
(.setDaemon runner daemon?)
;; 启动事件处理线程
(.start runner)
;; 返回一个实现了EventManager协议的实例
(reify
EventManager
;; add函数将"事件处理函数"添加到事件处理器的阻塞队列中
(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?判断事件处理线程是否处于等待状态
(waiting?
[this]
(or (Time/isThreadWaiting runner)
(= @processed @added)))
;; 关闭事件管理器
(shutdown
[this]
(reset! running false)
(.interrupt runner)
(.join runner)))))
)
"Error when processing an event")))))]
(.setDaemon runner daemon?)
;; 启动事件处理线程
(.start runner)
;; 返回一个实现了EventManager协议的实例
(reify
EventManager
;; add函数将"事件处理函数"添加到事件处理器的阻塞队列中
(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?判断事件处理线程是否处于等待状态
(waiting?
[this]
(or (Time/isThreadWaiting runner)
(= @processed @added)))
;; 关闭事件管理器
(shutdown
[this]
(reset! running false)
(.interrupt runner)
(.join runner)))))