storm操作zookeeper的主要函式都定義在名稱空間backtype.storm.cluster中（即cluster.clj檔案中）。 backtype.storm.cluster定義了兩個重要protocol：ClusterState和StormClusterState。clojure中的protocol可以看成java中的介面，封裝了一組方法。ClusterState協議中封裝了一組與zookeeper進行互動的基礎函式，如獲取子節點函式，獲取子節點資料函式等，ClusterState協議定義如下：

Zookeeper的操作 

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

Storm使用Zookeeper的操作 

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

Storm中在Zookeeper中儲存的目錄

(def ASSIGNMENTS-ROOT "assignments")
(def CODE-ROOT "code")
(def STORMS-ROOT "storms")
(def SUPERVISORS-ROOT "supervisors")
(def WORKERBEATS-ROOT "workerbeats")
(def ERRORS-ROOT "errors")

(def ASSIGNMENTS-SUBTREE (str "/" ASSIGNMENTS-ROOT))
(def STORMS-SUBTREE (str "/" STORMS-ROOT))
(def SUPERVISORS-SUBTREE (str "/" SUPERVISORS-ROOT))
(def WORKERBEATS-SUBTREE (str "/" WORKERBEATS-ROOT))
(def ERRORS-SUBTREE (str "/" ERRORS-ROOT))

從上面來看，在Zookeeper中主要是有如下的五個子目錄:

1. /assignments -> 任務分配資訊
2. /storms -> 正在執行的topology的ID
3. /supervisors -> 所有的Supervisors的心跳資訊
4. /workerbeats -> 所有的Worker的心跳
5. /errors -> 產生的出錯資訊

結構圖

/-{storm-zk-root}          -- storm在zookeeper上的根目錄(預設為/storm)
 |
 |-/assignments            -- topology的任務分配資訊
 |   |
 |   |-/{topology-id}      -- 這個下面儲存的是每個topology的assignments資訊包括： 對應的
 |                            nimbus上的程式碼目錄,所有task的啟動時間,每個task與機器、埠的對映
 |                            操作為(assignments)來獲取所有的assignments的值；
 |                            以及(assignment-info storm-id)來得到給定的storm-id對應的AssignmentInfo資訊
 |                            在AssignmentInfo中儲存的內容有:
 |                            :executor->node+port :executor->start-time-secs :node->host
 |                            具體定義在common.clj中的
 |                            (defrecord Assignment [master-code-dir node->host executor->node+port executor->start-time-secs])
 |
 |-/storms                 -- 這個目錄儲存所有正在執行的topology的id
 |   |                        
 |   |
 |   |-/{topology-id}      -- 這個檔案儲存這個topology的一些資訊，包括topology的
 |                            名字，topology開始執行的時間以及這個topology的狀態
 |                            操作(active-storms),獲得當前路徑下活躍的topology資料。儲存的內容參考類StormBase
 |                            (storm-base storm-id)得到給定的storm-id下的StormBase資料,具體定義在common.clj中的
 |                            (defrecord StormBase [storm-name launch-time-secs status num-workers component->executors])
 |
 |-/supervisors            -- 這個目錄儲存所有的supervisor的心跳資訊
 |   |                        
 |   |
 |   |-/{supervisor-id}    -- 這個檔案儲存的是supervisor的心跳資訊包括:心跳時間，主
 |                            機名，這個supervisor上worker的埠號執行時間(具體看SupervisorInfo類)
 |                            操作(supervisors)得到所有的supervisors節點
 |                            (supervisor-info supervisor-id)得到給定的supervisor-id對應的SupervisorInfo資訊
 |                            具體定義在common.clj中的
 |                            (defrecord SupervisorInfo [time-secs hostname assignment-id used-ports meta scheduler-meta uptime-secs])
 |
 |-/workerbeats                        -- 所有worker的心跳
 |   |
 |   |-/{topology-id}                  -- 這個目錄儲存這個topology的所有的worker的心跳資訊
 |       |
 |       |-/{supervisorId-port}        -- worker的心跳資訊，包括心跳的時
 |                                        間，worker執行時間以及一些統計資訊
 |                                        操作(heartbeat-storms)得到所有有心跳資料的topology，
 |                                        (get-worker-heartbeat storm-id node port)得到具體一個topology下的某個worker(node:port)的心跳狀況，
 |                                        (executor-beats storm-id executor->node+port)得到一個executor的心跳狀況
 |
 |-/errors                      -- 所有產生的error資訊
     |
     |-/{topology-id}           -- 這個目錄儲存這個topology下面的錯誤資訊
         |                         操作(error-topologies)得到出錯的topology
         |                         (errors storm-id component-id)得到給定的storm-id component-id下的出錯資訊
         |-/{component-id}