源码阅读之storm操作zookeeper-cluster.clj-6ren

源码阅读之storm操作zookeeper-cluster.clj

转载作者：qq735679552 更新时间：2022-09-28 22:32:09

CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.

这篇CFSDN的博客文章源码阅读之storm操作zookeeper-cluster.clj由作者收集整理，如果你对这篇文章有兴趣，记得点赞哟.

storm操作zookeeper的主要函数都定义在命名空间backtype.storm.cluster中（即cluster.clj文件中）。 backtype.storm.cluster定义了两个重要protocol：ClusterState和StormClusterState.

clojure中的protocol可以看成java中的接口，封装了一组方法。ClusterState协议中封装了一组与zookeeper进行交互的基础函数，如获取子节点函数，获取子节点数据函数等，ClusterState协议定义如下:

ClusterState协议。

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

。

StormClusterState协议封装了一组storm与zookeeper进行交互的函数，可以将StormClusterState协议中的函数看成ClusterState协议中函数的"组合"。StormClusterState协议定义如下:

StormClusterState协议。

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

命名空间backtype.storm.cluster除了定义ClusterState和StormClusterState这两个重要协议外，还定义了两个重要函数：mk-distributed-cluster-state和mk-storm-cluster-state.

mk-distributed-cluster-state函数如下:

该函数返回一个实现了ClusterState协议的对象，通过这个对象就可以与zookeeper进行交互了.

mk-distributed-cluster-state函数。

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

mk-storm-cluster-state函数定义如下:

mk-storm-cluster-state函数非常重要，该函数返回一个实现了StormClusterState协议的实例，通过该实例storm就可以更加方便与zookeeper进行交互.

在启动nimbus和supervisor的函数中均调用了mk-storm-cluster-state函数。关于nimbus和supervisor的启动将在之后的文章中介绍.

mk-storm-cluster-state函数。

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

zookeeper.clj中mk-client函数。

mk-client函数创建一个CuratorFramework实例，为该实例注册了CuratorListener，当一个后台操作完成或者指定的watch被触发时将会执行CuratorListener中的eventReceived()。eventReceived中调用的wacher函数就是mk-distributed-cluster-state中:watcher绑定的函数.

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

以上就是storm与zookeeper进行交互的源码分析，我觉得最重要的部分就是如何给zk client添加"wacher"，storm的很多功能都是通过zookeeper的wacher机制实现的，如"分配信息领取"。添加"wacher"大概分为以下几个步骤:

mk-distributed-cluster-state函数创建了一个zk client，并通过:watcher给该zk client指定了"wacher"函数，这个"wacher"函数只是简单调用ClusterState的callbacks集合中的函数，这样这个"wacher"函数执行哪些函数将由ClusterState实例决定 ClusterState实例提供register函数来更新callbacks集合，ClusterState实例被传递给了mk-storm-cluster-state函数，在mk-storm-cluster-state中调用register添加了一个函数(fn [type path] ... )，这个函数实现了"watcher"函数的全部逻辑 mk-storm-cluster-state中注册的函数执行的具体内容由StormClusterState实例决定，对zookeeper节点添加"观察"也是通过StormClusterState实例实现的，这样我们就可以通过StormClusterState实例对我们感兴趣的节点添加"观察"和"回调函数"，当节点或节点数据发生变化后，zk server就会给zk client发送"通知"，zk client中的"wather"函数将被调用，进而我们注册的"回到函数"将被执行.

总结。

这部分源码与zookeeper联系十分紧密，涉及了很多zookeeper中的概念和特性，如"数据观察"和"节点观察"等，有关zookeeper的wacher机制请参考：http://www.zzvips.com/article/130109.html，storm并没有直接使用zookeeper的api，而是使用Curator框架，Curator框架简化了访问zookeeper的操作。关于Curator框架请参考：http://www.zzvips.com/article/133166.html.

以上就是本文关于源码阅读之storm操作zookeeper-cluster.clj的全部内容了，希望对大家有所帮助。感谢各位的阅读！。

原文链接：http://blog.csdn.net/mangocream/article/details/42456901 。

最后此篇关于源码阅读之storm操作zookeeper-cluster.clj的文章就讲到这里了,如果你想了解更多关于源码阅读之storm操作zookeeper-cluster.clj的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！。

文章推荐：详解mybatis foreach collection示例

文章推荐： Kotlin基本类型自动装箱出现问题解决办法

文章推荐： VBS编程教程 (第1篇)

文章推荐： python+selenium打印当前页面的titl和url方法

Python Pandas 阅读
我正在尝试读取一个大型日志文件，该文件已使用不同的分隔符(遗留更改)进行了解析。此代码有效 import os, subprocess, time, re import pandas as pd f
Pascal(免费或快速)阅读
我试图理解在 Linux 下以 Turbo 模式(特别是 fpc -Mtp -vw)编译的 Free Pascal 中看到的有点神奇的行为。代码来自 Jack Crenshaw 的“让我们构建一个编译
阅读 R 中的表格？
我有一个具有以下结构的 txt 文件: NAME DATA1 DATA2 a 10 1,2,3 b 6 8,9 c 2
Pascal(免费或快速)阅读
我试图理解在 Linux 下以 Turbo 模式(特别是 fpc -Mtp -vw)编译的 Free Pascal 中看到的有点神奇的行为。代码来自 Jack Crenshaw 的“让我们构建一个编译
java - 阅读、理解和修复该代码
public class Bug1 { private String s; public void Bug1(){ s = "hello"; } public Stri
C#多线程文件IO(阅读)
我们有这样一种情况，我们的应用程序需要处理一系列文件，而不是同步执行此功能，我们希望采用多线程将工作负载分配给不同的线程。每一项工作是: 1.以只读方式打开文件 2.处理文件中的数据 3.将处理后的
c++ - 阅读(？)转义字符
我正在尝试读取 .php 文件并替换十六进制字符。php文件格式如下: 问题是它弄乱了转义字符 (\") 到目前为止我的代码: while(i=48 && str[i+2]=97 && str[i+
c# - 阅读 HTML？
我正在用 C# 开发一个程序，我需要一些帮助。我正在尝试创建一个数组或项目列表，显示在某个网站上。我想要做的是阅读 anchor 文本，它是 href。例如，这是 HTML:
android - 阅读/写作偏好是一项昂贵的操作吗？
我有一个偏好设置，它控制我的应用程序是否在用户单击按钮时播放声音(这种情况经常发生，想想计算器)。每次用户单击按钮时，都会调用以下方法: private void playButtonClickSou
ios - 阅读/查看标签末尾的更多信息
我正在尝试在我的标签末尾创建一个阅读更多按钮。我希望它默认显示 3 行。我正在用 swift 而不是 objective c 编写代码。只有当用户点击标签的阅读更多部分时，标签才会展开。它的外观和工作
c++ - 阅读/理解第三方代码
当您获得第三方库(c、c++)、开源(LGPL 说)但没有很好的文档时，了解它以便能够集成到您的应用程序中的最佳方法是什么？该库通常有一些示例程序，我最终使用 gdb 浏览了代码。还有其他建议/最佳
c - 阅读;线程安全与否？
同时从 2 个或更多不同线程对同一个文件描述符使用 pread 是否有问题？最佳答案 pread 本身是线程安全的，因为它不在 list of unsafe functions 上.所以调用它是安全
python - 阅读 Pandas 数据框时跳过包含特定值的特定行
当您使用命令 pd.read_csv 读取 csv 时，如何跳过连续包含特定值的行？如果在第 50、55 行，第一列的值为 100，那么我想在读取 csv 文件时跳过这些行。我如何将这些命令放入像 p
c# - 阅读 T4 文件的内容我缺少什么？
我迫切需要在 C# 中使用 T4 生成 HTML 输出。我正在使用 Runtime-T4-Files 并选择“TextTemplatingFilePreprocessor”而不是“TextTempl
sap - ERP 阅读/练习什么？
今年夏天我在实习期间一直在学习 ERP 应用程序。由于我是一名即将毕业的程序员，我希望有一个可靠的软件分支可以帮助我完成工作，直到我确定下一步该做什么(直到我对大局有一个很好的了解)。到现在为止，我刚
scala - 阅读 Parquet 时是否有可能保持列顺序？
将包含列(例如“a”、“b”)的数据帧保存为 parquet，然后在稍后的时间点读取 parquet 不会提供相同的列顺序(可能是“b”、“a”fe)文件保存为。不幸的是，我无法弄清楚订单是如何受到
Swift - 阅读 Google 表格
我正在开发一个使用谷歌表格作为数据库的应用程序，但我不知道如何让 Swift 从谷歌表格中读取。我浏览了 API 网站和一些问题，但刚开始我需要一些帮助。到目前为止，我有；私有(private)让范
java - 阅读 Swing 是否值得？
我打算阅读swing concept，如果值得一读，请推荐一些学习 Material 最佳答案自 AWT 崩溃以来，Java 的 GUI 工具包太多了。即使是 Swing 也被评论家严重低估，但他们
j - 阅读 J 代码的最佳策略
我已经使用 J 几个月了，我发现阅读不熟悉的代码(例如，不是我自己写的)是该语言最具挑战性的方面之一，尤其是在默认情况下。过了一会儿，我想出了这个策略: 1)将代码段复制到word文档中 2)从(1)
iphone - 阅读 iPhone 短信？
很难说出这里问的是什么。这个问题是含糊的、模糊的、不完整的、过于宽泛的或修辞性的，无法以目前的形式得到合理的回答。如需帮助澄清此问题以便重新打开它，visit the help center 。已关

qq735679552

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

源码阅读之storm操作zookeeper-cluster.clj