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定义或声明的情况下, 方便的操作他们