(defprotocol ClusterState  (set-ephemeral-node [this path data])  (delete-node [this path])  (create-sequential [this path data])  ;; if node does not exist, create persistent with this data  (set-data [this path data])  (get-data [this path watch?])  (get-version [this path watch?])  (get-data-with-version [this path watch?])  (get-children [this path watch?])  (mkdirs [this path])  (close [this])  (register [this callback])  (unregister [this id]))

(defprotocol StormClusterState (assignments [this callback]) (assignment-info [this storm-id callback]) (assignment-info-with-version [this storm-id callback]) (assignment-version [this storm-id callback]) (active-storms [this]) (storm-base [this storm-id callback]) (get-worker-heartbeat [this storm-id node port]) (executor-beats [this storm-id executor->node+port]) (supervisors [this callback]) (supervisor-info [this supervisor-id]) ;; returns nil if doesn't exist (setup-heartbeats! [this storm-id]) (teardown-heartbeats! [this storm-id]) (teardown-topology-errors! [this storm-id]) (heartbeat-storms [this]) (error-topologies [this]) (worker-heartbeat! [this storm-id node port info]) (remove-worker-heartbeat! [this storm-id node port]) (supervisor-heartbeat! [this supervisor-id info]) (activate-storm! [this storm-id storm-base]) (update-storm! [this storm-id new-elems]) (remove-storm-base! [this storm-id]) (set-assignment! [this storm-id info]) (remove-storm! [this storm-id]) (report-error [this storm-id task-id node port error]) (errors [this storm-id task-id]) (disconnect [this]))

(defn mk-distributed-cluster-state;; conf綁定了storm.yaml中的配置信息，是一個map對象[conf];; zk綁定一個zk client，Storm使用CuratorFramework與Zookeeper進行交互 (let [zk (zk/mk-client conf (conf STORM-ZOOKEEPER-SERVERS) (conf STORM-ZOOKEEPER-PORT)            :auth-conf conf)]  ;; 創建storm集群在zookeeper上的根目錄，默認值為/storm  (zk/mkdirs zk (conf STORM-ZOOKEEPER-ROOT))  (.close zk));; callbacks綁定回調函數集合，是一個map對象(let [callbacks (atom {})  ;; active標示zookeeper集群狀態   active (atom true)  ;; zk重新綁定新的zk client，該zk client設置了watcher，這樣當zookeeper集群的狀態發生變化時，zk server會給zk client發送相應的event，zk client設置的watcher會調用callbacks中相應回調函數來處理event  ;; 啟動nimbus時，callbacks是一個空集合，所以nimbus端收到event后不會調用任何回調函數；但是啟動supervisor時，callbacks中注冊了回調函數，所以當supervisor收到zk server發送的event后，會調用相應的回調函數  ;; mk-client函數定義在zookeeper.clj文件中，請參見其定義部分  zk (zk/mk-client conf           (conf STORM-ZOOKEEPER-SERVERS)           (conf STORM-ZOOKEEPER-PORT)           :auth-conf conf           :root (conf STORM-ZOOKEEPER-ROOT)           ;; :watcher綁定一個函數，指定zk client的默認watcher函數，state標示當前zk client的狀態；type標示事件類型；path標示zookeeper上產生該事件的znode           ;; 該watcher函數主要功能就是執行callbacks集合中的函數，callbacks集合中的函數是在mk-storm-cluster-state函數中通過調用ClusterState的register函數添加的           :watcher (fn [state type path]                (when @active                 (when-not (= :connected state)                  (log-warn "Received event " state ":" type ":" path " with disconnected Zookeeper."))                 (when-not (= :none type)                  (doseq [callback (vals @callbacks)]                   (callback type path))))))];; reify相當于java中的implements，這里表示實現一個協議(reify ClusterState ;; register函數用于將回調函數加入callbacks中，key是一個32位的標識 (register  [this callback]  (let [id (uuid)]   (swap! callbacks assoc id callback)   id)) ;; unregister函數用于將指定key的回調函數從callbacks中刪除 (unregister  [this id]  (swap! callbacks dissoc id)) ;; 在zookeeper上添加一個臨時節點 (set-ephemeral-node  [this path data]  (zk/mkdirs zk (parent-path path))  (if (zk/exists zk path false)   (try-cause    (zk/set-data zk path data) ; should verify that it's ephemeral    (catch KeeperException$NoNodeException e     (log-warn-error e "Ephemeral node disappeared between checking for existing and setting data")     (zk/create-node zk path data :ephemeral)     ))   (zk/create-node zk path data :ephemeral))) ;; 在zookeeper上添加一個順序節點 (create-sequential  [this path data]  (zk/create-node zk path data :sequential)) ;; 修改某個節點數據 (set-data  [this path data]  ;; note: this does not turn off any existing watches  (if (zk/exists zk path false)   (zk/set-data zk path data)   (do    (zk/mkdirs zk (parent-path path))    (zk/create-node zk path data :persistent)))) ;; 刪除指定節點 (delete-node  [this path]  (zk/delete-recursive zk path)) ;; 獲取指定節點數據。path標示節點路徑；watch?是一個布爾類型值，表示是否需要對該節點進行"觀察"，如果watch?=true，當調用set-data函數修改該節點數據后， ;; 會給zk client發送一個事件，zk client接收事件后，會調用創建zk client時指定的默認watcher函數（即:watcher綁定的函數） (get-data  [this path watch?]  (zk/get-data zk path watch?)) ;; 與get-data函數的區別就是獲取指定節點數據的同時，獲取節點數據的version，version表示節點數據修改的次數 (get-data-with-version  [this path watch?]  (zk/get-data-with-version zk path watch?)) ;; 獲取指定節點的version，watch?的含義與get-data函數中的watch?相同 (get-version   [this path watch?]  (zk/get-version zk path watch?)) ;; 獲取指定節點的子節點列表，watch?的含義與get-data函數中的watch?相同 (get-children  [this path watch?]  (zk/get-children zk path watch?)) ;; 在zookeeper上創建一個節點 (mkdirs  [this path]  (zk/mkdirs zk path)) ;; 關閉zk client (close  [this]  (reset! active false)  (.close zk)))))

(defn mk-storm-cluster-state [cluster-state-spec] ;; satisfies?謂詞相當于java中的instanceof，判斷cluster-state-spec是不是ClusterState實例 (let [[solo? cluster-state] (if (satisfies? ClusterState cluster-state-spec)              [false cluster-state-spec]              [true (mk-distributed-cluster-state cluster-state-spec)])  ;; 綁定topology id->回調函數的map，當/assignments/{topology id}數據發生變化時，zk client執行assignment-info-callback中topology id所對應的回調函數  assignment-info-callback (atom {})  ;; assignment-info-with-version-callback與assignment-info-callback類似  assignment-info-with-version-callback (atom {})  ;; assignment-version-callback與assignments-callback類似  assignment-version-callback (atom {})  ;; 當/supervisors標示的znode的子節點發生變化時，zk client執行supervisors-callback指向的函數  supervisors-callback (atom nil)  ;; 當/assignments標示的znode的子節點發生變化時，zk client執行assignments-callback指向的函數  assignments-callback (atom nil)  ;; 當/storms/{topology id}標示的znode的數據發生變化時，zk client執行storm-base-callback中topology id所對應的回調函數  storm-base-callback (atom {})  ;; register函數將"回調函數(fn ...)"添加到cluster-state的callbacks集合中，并返回標示該回調函數的uuid  state-id (register        cluster-state        ;; 定義"回調函數"，type標示事件類型，path標示znode        (fn [type path]         ;; subtree綁定路徑前綴如"assignments"、"storms"、"supervisors"等，args存放topology id         (let [[subtree & args] (tokenize-path path)]          ;; condp相當于java中的switch          (condp = subtree           ;; 當subtree="assignments"時，如果args為空，說明是/assignments的子節點發生變化，執行assignments-callback指向的回調函數，否則     ;; 說明/assignments/{topology id}標示的節點數據發生變化，執行assignment-info-callback指向的回調函數           ASSIGNMENTS-ROOT (if (empty? args)                    (issue-callback! assignments-callback)                    (issue-map-callback! assignment-info-callback (first args)))           ;; 當subtree="supervisors"時，說明是/supervisors的子節點發生變化，執行supervisors-callback指向的回調函數           SUPERVISORS-ROOT (issue-callback! supervisors-callback)           ;; 當subtree="storms"時，說明是/storms/{topology id}標示的節點數據發生變化，執行storm-base-callback指向的回調函數           STORMS-ROOT (issue-map-callback! storm-base-callback (first args))           ;; this should never happen           (exit-process! 30 "Unknown callback for subtree " subtree args)))))];; 在zookeeper上創建storm運行topology所必需的znode(doseq [p [ASSIGNMENTS-SUBTREE STORMS-SUBTREE SUPERVISORS-SUBTREE WORKERBEATS-SUBTREE ERRORS-SUBTREE]] (mkdirs cluster-state p));; 返回一個實現StormClusterState協議的實例(reify StormClusterState ;; 獲取/assignments的子節點列表，如果callback不為空，將其賦值給assignments-callback，并對/assignments添加"節點觀察" (assignments  [this callback]  (when callback   (reset! assignments-callback callback))  (get-children cluster-state ASSIGNMENTS-SUBTREE (not-nil? callback))) ;; 獲取/assignments/{storm-id}節點數據，即storm-id的分配信息，如果callback不為空，將其添加到assignment-info-callback中，并對/assignments/{storm-id}添加"數據觀察" (assignment-info  [this storm-id callback]  (when callback   (swap! assignment-info-callback assoc storm-id callback))  (maybe-deserialize (get-data cluster-state (assignment-path storm-id) (not-nil? callback)))) ;; 獲取/assignments/{storm-id}節點數據包括version信息，如果callback不為空，將其添加到assignment-info-with-version-callback中，并對/assignments/{storm-id}添加"數據觀察" (assignment-info-with-version   [this storm-id callback]  (when callback   (swap! assignment-info-with-version-callback assoc storm-id callback))  (let [{data :data version :version}      (get-data-with-version cluster-state (assignment-path storm-id) (not-nil? callback))]  {:data (maybe-deserialize data)   :version version})) ;; 獲取/assignments/{storm-id}節點數據的version信息，如果callback不為空，將其添加到assignment-version-callback中，并對/assignments/{storm-id}添加"數據觀察" (assignment-version   [this storm-id callback]  (when callback   (swap! assignment-version-callback assoc storm-id callback))  (get-version cluster-state (assignment-path storm-id) (not-nil? callback))) ;; 獲取storm集群中正在運行的topology id即/storms的子節點列表 (active-storms  [this]  (get-children cluster-state STORMS-SUBTREE false)) ;; 獲取storm集群中所有有心跳的topology id即/workerbeats的子節點列表 (heartbeat-storms  [this]  (get-children cluster-state WORKERBEATS-SUBTREE false)) ;; 獲取所有有錯誤的topology id即/errors的子節點列表 (error-topologies  [this]  (get-children cluster-state ERRORS-SUBTREE false)) ;; 獲取指定storm-id進程的心跳信息，即/workerbeats/{storm-id}/{node-port}節點數據 (get-worker-heartbeat  [this storm-id node port]  (-> cluster-state    (get-data (workerbeat-path storm-id node port) false)    maybe-deserialize)) ;; 獲取指定進程中所有線程的心跳信息 (executor-beats  [this storm-id executor->node+port]  ;; need to take executor->node+port in explicitly so that we don't run into a situation where a  ;; long dead worker with a skewed clock overrides all the timestamps. By only checking heartbeats  ;; with an assigned node+port, and only reading executors from that heartbeat that are actually assigned,  ;; we avoid situations like that  (let [node+port->executors (reverse-map executor->node+port)     all-heartbeats (for [[[node port] executors] node+port->executors]              (->> (get-worker-heartbeat this storm-id node port)                (convert-executor-beats executors)                ))]   (apply merge all-heartbeats))) ;; 獲取/supervisors的子節點列表，如果callback不為空，將其賦值給supervisors-callback，并對/supervisors添加"節點觀察"  (supervisors  [this callback]  (when callback   (reset! supervisors-callback callback))  (get-children cluster-state SUPERVISORS-SUBTREE (not-nil? callback))) ;; 獲取/supervisors/{supervisor-id}節點數據，即supervisor的心跳信息 (supervisor-info  [this supervisor-id]  (maybe-deserialize (get-data cluster-state (supervisor-path supervisor-id) false))) ;; 設置進程心跳信息 (worker-heartbeat!  [this storm-id node port info]  (set-data cluster-state (workerbeat-path storm-id node port) (Utils/serialize info))) ;; 刪除進程心跳信息 (remove-worker-heartbeat!  [this storm-id node port]  (delete-node cluster-state (workerbeat-path storm-id node port))) ;; 創建指定storm-id的topology的用于存放心跳信息的節點 (setup-heartbeats!  [this storm-id]  (mkdirs cluster-state (workerbeat-storm-root storm-id))) ;; 刪除指定storm-id的topology的心跳信息節點 (teardown-heartbeats!  [this storm-id]  (try-cause   (delete-node cluster-state (workerbeat-storm-root storm-id))   (catch KeeperException e    (log-warn-error e "Could not teardown heartbeats for " storm-id)))) ;; 刪除指定storm-id的topology的錯誤信息節點 (teardown-topology-errors!  [this storm-id]  (try-cause   (delete-node cluster-state (error-storm-root storm-id))   (catch KeeperException e    (log-warn-error e "Could not teardown errors for " storm-id)))) ;; 創建臨時節點存放supervisor的心跳信息 (supervisor-heartbeat!  [this supervisor-id info]  (set-ephemeral-node cluster-state (supervisor-path supervisor-id) (Utils/serialize info))) ;; 創建/storms/{storm-id}節點 (activate-storm!  [this storm-id storm-base]  (set-data cluster-state (storm-path storm-id) (Utils/serialize storm-base))) ;; 更新topology對應的StormBase對象，即更新/storm/{storm-id}節點 (update-storm!  [this storm-id new-elems]  ;; base綁定storm-id在zookeeper上的StormBase對象  (let [base (storm-base this storm-id nil)     ;; executors綁定component名稱->組件并行度的map     executors (:component->executors base)     ;; new-elems綁定合并后的組件并行度map，update函數將組件新并行度map合并到舊map中     new-elems (update new-elems :component->executors (partial merge executors))]   ;; 更新StormBase對象中的組件并行度map，并寫入zookeeper的/storms/{storm-id}節點   (set-data cluster-state (storm-path storm-id)        (-> base          (merge new-elems)          Utils/serialize)))) ;; 獲取storm-id的StormBase對象，即讀取/storms/{storm-id}節點數據，如果callback不為空，將其賦值給storm-base-callback，并為/storms/{storm-id}節點添加"數據觀察" (storm-base  [this storm-id callback]  (when callback   (swap! storm-base-callback assoc storm-id callback))  (maybe-deserialize (get-data cluster-state (storm-path storm-id) (not-nil? callback)))) ;; 刪除storm-id的StormBase對象，即刪除/storms/{storm-id}節點 (remove-storm-base!  [this storm-id]  (delete-node cluster-state (storm-path storm-id))) ;; 更新storm-id的分配信息，即更新/assignments/{storm-id}節點數據 (set-assignment!  [this storm-id info]  (set-data cluster-state (assignment-path storm-id) (Utils/serialize info))) ;; 刪除storm-id的分配信息，同時刪除其StormBase信息，即刪除/assignments/{storm-id}節點和/storms/{storm-id}節點 (remove-storm!  [this storm-id]  (delete-node cluster-state (assignment-path storm-id))  (remove-storm-base! this storm-id)) ;; 將組件異常信息寫入zookeeper (report-error  [this storm-id component-id node port error]  ;; path綁定"/errors/{storm-id}/{component-id}"  (let [path (error-path storm-id component-id)     ;; data綁定異常信息，包括異常時間、異常堆棧信息、主機和端口     data {:time-secs (current-time-secs) :error (stringify-error error) :host node :port port}     ;; 創建/errors/{storm-id}/{component-id}節點     _ (mkdirs cluster-state path)     ;; 創建/errors/{storm-id}/{component-id}的子順序節點，并寫入異常信息     _ (create-sequential cluster-state (str path "/e") (Utils/serialize data))     ;; to-kill綁定除去順序節點編號最大的前10個節點的剩余節點的集合     to-kill (->> (get-children cluster-state path false)            (sort-by parse-error-path)            reverse            (drop 10))]   ;; 刪除to-kill中包含的節點   (doseq [k to-kill]    (delete-node cluster-state (str path "/" k))))) ;; 得到給定的storm-id component-id下的異常信息 (errors  [this storm-id component-id]  (let [path (error-path storm-id component-id)     _ (mkdirs cluster-state path)     children (get-children cluster-state path false)     errors (dofor [c children]            (let [data (-> (get-data cluster-state (str path "/" c) false)                    maybe-deserialize)]             (when data              (struct TaskError (:error data) (:time-secs data) (:host data) (:port data))              )))     ]   (->> (filter not-nil? errors)      (sort-by (comp - :time-secs))))) ;; 關閉連接，在關閉連接前，將回調函數從cluster-state的callbacks中刪除 (disconnect  [this]  (unregister cluster-state state-id)  (when solo?   (close cluster-state))))))

(defnk mk-client [conf servers port  :root ""  :watcher default-watcher  :auth-conf nil] (let [fk (Utils/newCurator conf servers port root (when auth-conf (ZookeeperAuthInfo. auth-conf)))]  (.. fk    (getCuratorListenable)    (addListener     (reify CuratorListener      (^void eventReceived [this ^CuratorFramework _fk ^CuratorEvent e]        (when (= (.getType e) CuratorEventType/WATCHED)         (let [^WatchedEvent event (.getWatchedEvent e)]          (watcher (zk-keeper-states (.getState event))              (zk-event-types (.getType event))              (.getPath event))))))))  (.start fk)  fk))