storm操作zookeeper源码分析-cluster.clj

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

(defprotocol StormClusterState
 (assignments[thiscallback])
 (assignment-info[thisstorm-idcallback])
 (assignment-info-with-version[thisstorm-idcallback])
 (assignment-version[thisstorm-idcallback])
 (active-storms[this])
 (storm-base[thisstorm-idcallback])
 (get-worker-heartbeat[thisstorm-idnode port])
 (executor-beats[thisstorm-idexecutor->node+port])
 (supervisors[thiscallback])
 (supervisor-info[thissupervisor-id]);; returns nil if doesn‘t exist
 (setup-heartbeats![thisstorm-id])
 (teardown-heartbeats![thisstorm-id])
 (teardown-topology-errors![thisstorm-id])
 (heartbeat-storms[this])
 (error-topologies[this])
 (worker-heartbeat![thisstorm-idnode portinfo])
 (remove-worker-heartbeat![thisstorm-idnode port])
 (supervisor-heartbeat![thissupervisor-idinfo])
 (activate-storm![thisstorm-idstorm-base])
 (update-storm![thisstorm-idnew-elems])
 (remove-storm-base![thisstorm-id])
 (set-assignment![thisstorm-idinfo])
 (remove-storm![thisstorm-id])
 (report-error[thisstorm-idtask-idnode porterror])
 (errors[thisstorm-idtask-id])
 (disconnect[this]))

(defn mk-distributed-cluster-state
 ;; conf绑定了storm.yaml中的配置信息，是一个map对象
 [conf]
 ;; zk绑定一个zk client，Storm使用CuratorFramework与Zookeeper进行交互
 (let [zk (zk/mk-clientconf (confSTORM-ZOOKEEPER-SERVERS) (confSTORM-ZOOKEEPER-PORT) :auth-confconf)]
   ;; 创建storm集群在zookeeper上的根目录，默认值为/storm
   (zk/mkdirszk (confSTORM-ZOOKEEPER-ROOT))
   (.closezk))
 ;; callbacks绑定回调函数集合，是一个map对象
 (let [callbacks (atom{})
       ;; active标示zookeeper集群状态
       active (atomtrue)
       ;; 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-clientconf
                        (confSTORM-ZOOKEEPER-SERVERS)
                        (confSTORM-ZOOKEEPER-PORT)
                        :auth-confconf
                        :root (confSTORM-ZOOKEEPER-ROOT)
                        ;; :watcher绑定一个函数，指定zk client的默认watcher函数，state标示当前zk client的状态；type标示事件类型；path标示zookeeper上产生该事件的znode
                        ;; 该watcher函数主要功能就是执行callbacks集合中的函数，callbacks集合中的函数是在mk-storm-cluster-state函数中通过调用ClusterState的register函数添加的
                        :watcher (fn [statetypepath]
                                   (when @active
                                     (when-not (= :connectedstate)
                                       (log-warn"Received event "state":"type":"path " with disconnected Zookeeper."))
                                     (when-not (= :nonetype)
                                       (doseq [callback (vals @callbacks)]
                                         (callbacktypepath))))))]
   ;; reify相当于java中的implements，这里表示实现一个协议
   (reify
    ClusterState
    ;; register函数用于将回调函数加入callbacks中，key是一个32位的标识
    (register
      [thiscallback]
      (let [id (uuid)]
        (swap!callbacksassoc idcallback)
        id))
    ;; unregister函数用于将指定key的回调函数从callbacks中删除
    (unregister
      [thisid]
      (swap!callbacksdissoc id))
    ;; 在zookeeper上添加一个临时节点
    (set-ephemeral-node
      [thispath data]
      (zk/mkdirszk (parent-pathpath))
      (if (zk/existszkpath false)
        (try-cause
          (zk/set-datazkpath data) ; should verify that it‘s ephemeral
          (catchKeeperException$NoNodeExceptione
            (log-warn-errore"Ephemeral node disappeared between checking for existing and setting data")
            (zk/create-nodezkpath data:ephemeral)
            ))
        (zk/create-nodezkpath data:ephemeral)))
    ;; 在zookeeper上添加一个顺序节点
    (create-sequential
      [thispath data]
      (zk/create-nodezkpath data:sequential))
    ;; 修改某个节点数据
    (set-data
      [thispath data]
      ;; note: this does not turn off any existing watches
      (if (zk/existszkpath false)
        (zk/set-datazkpath data)
        (do
          (zk/mkdirszk (parent-pathpath))
          (zk/create-nodezkpath data:persistent))))
    ;; 删除指定节点
    (delete-node
      [thispath]
      (zk/delete-recursivezkpath))
    ;; 获取指定节点数据。path标示节点路径；watch?是一个布尔类型值，表示是否需要对该节点进行"观察"，如果watch?=true，当调用set-data函数修改该节点数据后，
    ;; 会给zk client发送一个事件，zk client接收事件后，会调用创建zk client时指定的默认watcher函数（即:watcher绑定的函数）
    (get-data
      [thispath watch?]
      (zk/get-datazkpath watch?))
    ;; 与get-data函数的区别就是获取指定节点数据的同时，获取节点数据的version，version表示节点数据修改的次数
    (get-data-with-version
      [thispath watch?]
      (zk/get-data-with-versionzkpath watch?))
    ;; 获取指定节点的version，watch?的含义与get-data函数中的watch?相同
    (get-version
      [thispath watch?]
      (zk/get-versionzkpath watch?))
    ;; 获取指定节点的子节点列表，watch?的含义与get-data函数中的watch?相同
    (get-children
      [thispath watch?]
      (zk/get-childrenzkpath watch?))
    ;; 在zookeeper上创建一个节点
    (mkdirs
      [thispath]
      (zk/mkdirszkpath))
    ;; 关闭zk client
    (close
      [this]
      (reset!activefalse)
      (.closezk)))))

(defn mk-storm-cluster-state
 [cluster-state-spec]
 ;; satisfies?谓词相当于java中的instanceof，判断cluster-state-spec是不是ClusterState实例
 (let [[solo?cluster-state] (if (satisfies?ClusterStatecluster-state-spec)
                               [falsecluster-state-spec]
                               [true (mk-distributed-cluster-statecluster-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 (atomnil)
   　　;; 当/assignments标示的znode的子节点发生变化时，zk client执行assignments-callback指向的函数
       assignments-callback (atomnil)
  　　 ;; 当/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 [typepath]
            　　　　 ;; subtree绑定路径前缀如"assignments"、"storms"、"supervisors"等，args存放topology id
                    (let [[subtree&args] (tokenize-pathpath)]
              　　　　 ;; 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 "subtreeargs)))))]
   ;; 在zookeeper上创建storm运行topology所必需的znode
   (doseq [p [ASSIGNMENTS-SUBTREESTORMS-SUBTREESUPERVISORS-SUBTREEWORKERBEATS-SUBTREEERRORS-SUBTREE]]
     (mkdirscluster-state p))
   ;; 返回一个实现StormClusterState协议的实例
   (reify
     StormClusterState
     ;; 获取/assignments的子节点列表，如果callback不为空，将其赋值给assignments-callback，并对/assignments添加"节点观察"
     (assignments
       [thiscallback]
       (when callback
         (reset!assignments-callbackcallback))
       (get-childrencluster-stateASSIGNMENTS-SUBTREE (not-nil?callback)))
     ;; 获取/assignments/{storm-id}节点数据，即storm-id的分配信息，如果callback不为空，将其添加到assignment-info-callback中，并对/assignments/{storm-id}添加"数据观察"
     (assignment-info
       [thisstorm-idcallback]
       (when callback
         (swap!assignment-info-callbackassoc storm-idcallback))
       (maybe-deserialize (get-datacluster-state (assignment-pathstorm-id) (not-nil?callback))))
     ;; 获取/assignments/{storm-id}节点数据包括version信息，如果callback不为空，将其添加到assignment-info-with-version-callback中，并对/assignments/{storm-id}添加"数据观察"
     (assignment-info-with-version
       [thisstorm-idcallback]
       (when callback
         (swap!assignment-info-with-version-callbackassoc storm-idcallback))
       (let [{data:dataversion:version}
             (get-data-with-versioncluster-state (assignment-pathstorm-id) (not-nil?callback))]
       {:data (maybe-deserializedata)
        :versionversion}))
     ;; 获取/assignments/{storm-id}节点数据的version信息，如果callback不为空，将其添加到assignment-version-callback中，并对/assignments/{storm-id}添加"数据观察"
     (assignment-version
       [thisstorm-idcallback]
       (when callback
         (swap!assignment-version-callbackassoc storm-idcallback))
       (get-versioncluster-state (assignment-pathstorm-id) (not-nil?callback)))
     ;; 获取storm集群中正在运行的topology id即/storms的子节点列表
     (active-storms
       [this]
       (get-childrencluster-stateSTORMS-SUBTREEfalse))
     ;; 获取storm集群中所有有心跳的topology id即/workerbeats的子节点列表
     (heartbeat-storms
       [this]
       (get-childrencluster-stateWORKERBEATS-SUBTREEfalse))
     ;; 获取所有有错误的topology id即/errors的子节点列表
     (error-topologies
       [this]
       (get-childrencluster-stateERRORS-SUBTREEfalse))
     ;; 获取指定storm-id进程的心跳信息，即/workerbeats/{storm-id}/{node-port}节点数据
     (get-worker-heartbeat
       [thisstorm-idnode port]
       (-> cluster-state
           (get-data (workerbeat-pathstorm-idnode port) false)
           maybe-deserialize))
     ;; 获取指定进程中所有线程的心跳信息
     (executor-beats
       [thisstorm-idexecutor->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-mapexecutor->node+port)
             all-heartbeats (for [[[node port]executors]node+port->executors]
                              (->> (get-worker-heartbeatthisstorm-idnode port)
                                   (convert-executor-beatsexecutors)
                                   ))]
         (apply merge all-heartbeats)))
     ;; 获取/supervisors的子节点列表，如果callback不为空，将其赋值给supervisors-callback，并对/supervisors添加"节点观察"
     (supervisors
       [thiscallback]
       (when callback
         (reset!supervisors-callbackcallback))
       (get-childrencluster-stateSUPERVISORS-SUBTREE (not-nil?callback)))
     ;; 获取/supervisors/{supervisor-id}节点数据，即supervisor的心跳信息
     (supervisor-info
       [thissupervisor-id]
       (maybe-deserialize (get-datacluster-state (supervisor-pathsupervisor-id) false)))
     ;; 设置进程心跳信息
     (worker-heartbeat!
       [thisstorm-idnode portinfo]
       (set-datacluster-state (workerbeat-pathstorm-idnode port) (Utils/serializeinfo)))
     ;; 删除进程心跳信息
     (remove-worker-heartbeat!
       [thisstorm-idnode port]
       (delete-nodecluster-state (workerbeat-pathstorm-idnode port)))
     ;; 创建指定storm-id的topology的用于存放心跳信息的节点
     (setup-heartbeats!
       [thisstorm-id]
       (mkdirscluster-state (workerbeat-storm-rootstorm-id)))
     ;; 删除指定storm-id的topology的心跳信息节点
     (teardown-heartbeats!
       [thisstorm-id]
       (try-cause
         (delete-nodecluster-state (workerbeat-storm-rootstorm-id))
         (catchKeeperExceptione
           (log-warn-errore"Could not teardown heartbeats for "storm-id))))
     ;; 删除指定storm-id的topology的错误信息节点
     (teardown-topology-errors!
       [thisstorm-id]
       (try-cause
         (delete-nodecluster-state (error-storm-rootstorm-id))
         (catchKeeperExceptione
           (log-warn-errore"Could not teardown errors for "storm-id))))
     ;; 创建临时节点存放supervisor的心跳信息
     (supervisor-heartbeat!
       [thissupervisor-idinfo]
       (set-ephemeral-nodecluster-state (supervisor-pathsupervisor-id) (Utils/serializeinfo)))
     ;; 创建/storms/{storm-id}节点
     (activate-storm!
       [thisstorm-idstorm-base]
       (set-datacluster-state (storm-pathstorm-id) (Utils/serializestorm-base)))
     ;; 更新topology对应的StormBase对象，即更新/storm/{storm-id}节点
     (update-storm!
       [thisstorm-idnew-elems]
       ;; base绑定storm-id在zookeeper上的StormBase对象
       (let [base (storm-basethisstorm-idnil)
             ;; executors绑定component名称->组件并行度的map
             executors (:component->executorsbase)
             ;; new-elems绑定合并后的组件并行度map，update函数将组件新并行度map合并到旧map中
             new-elems (updatenew-elems:component->executors (partial merge executors))]
         ;; 更新StormBase对象中的组件并行度map，并写入zookeeper的/storms/{storm-id}节点
         (set-datacluster-state (storm-pathstorm-id)
                   (-> base
                       (merge new-elems)
                       Utils/serialize))))
     ;; 获取storm-id的StormBase对象，即读取/storms/{storm-id}节点数据，如果callback不为空，将其赋值给storm-base-callback，并为/storms/{storm-id}节点添加"数据观察"
     (storm-base
       [thisstorm-idcallback]
       (when callback
         (swap!storm-base-callbackassoc storm-idcallback))
       (maybe-deserialize (get-datacluster-state (storm-pathstorm-id) (not-nil?callback))))
     ;; 删除storm-id的StormBase对象，即删除/storms/{storm-id}节点
     (remove-storm-base!
       [thisstorm-id]
       (delete-nodecluster-state (storm-pathstorm-id)))
     ;; 更新storm-id的分配信息，即更新/assignments/{storm-id}节点数据
     (set-assignment!
       [thisstorm-idinfo]
       (set-datacluster-state (assignment-pathstorm-id) (Utils/serializeinfo)))
     ;; 删除storm-id的分配信息，同时删除其StormBase信息，即删除/assignments/{storm-id}节点和/storms/{storm-id}节点
     (remove-storm!
       [thisstorm-id]
       (delete-nodecluster-state (assignment-pathstorm-id))
       (remove-storm-base!thisstorm-id))
     ;; 将组件异常信息写入zookeeper
     (report-error
       [thisstorm-idcomponent-idnode porterror]
       ;; path绑定"/errors/{storm-id}/{component-id}"
       (let [path (error-pathstorm-idcomponent-id)
             ;; data绑定异常信息，包括异常时间、异常堆栈信息、主机和端口
             data{:time-secs (current-time-secs) :error (stringify-errorerror) :hostnode :portport}
             ;; 创建/errors/{storm-id}/{component-id}节点
             _ (mkdirscluster-statepath)
             ;; 创建/errors/{storm-id}/{component-id}的子顺序节点，并写入异常信息
             _ (create-sequentialcluster-state (str path "/e") (Utils/serializedata))
             ;; to-kill绑定除去顺序节点编号最大的前10个节点的剩余节点的集合
             to-kill (->> (get-childrencluster-statepath false)
                          (sort-by parse-error-path)
                          reverse
                          (drop 10))]
         ;; 删除to-kill中包含的节点
         (doseq [kto-kill]
           (delete-nodecluster-state (str path "/"k)))))
     ;; 得到给定的storm-id component-id下的异常信息
     (errors
       [thisstorm-idcomponent-id]
       (let [path (error-pathstorm-idcomponent-id)
             _ (mkdirscluster-statepath)
             children (get-childrencluster-statepath false)
             errors (dofor[cchildren]
                           (let [data (-> (get-datacluster-state (str path "/"c) false)
                                          maybe-deserialize)]
                             (when data
                               (struct TaskError (:errordata) (:time-secsdata) (:hostdata) (:portdata))
                               )))
             ]
         (->> (filter not-nil?errors)
              (sort-by (comp - :time-secs)))))
     ;; 关闭连接，在关闭连接前，将回调函数从cluster-state的callbacks中删除
     (disconnect
       [this]
       (unregistercluster-statestate-id)
       (when solo?
         (closecluster-state))))))

(defnkmk-client
 [confserversport
  :root""
  :watcherdefault-watcher
  :auth-confnil]
 (let [fk (Utils/newCuratorconfserversportroot (when auth-conf (ZookeeperAuthInfo.auth-conf)))]
   (.. fk
       (getCuratorListenable)
       (addListener
         (reifyCuratorListener
           (^voideventReceived[this^CuratorFramework_fk^CuratorEvente]
                  (when (= (.getTypee) CuratorEventType/WATCHED)
                    (let [^WatchedEventevent (.getWatchedEvente)]
                      (watcher (zk-keeper-states (.getStateevent))
                               (zk-event-types (.getTypeevent))
                               (.getPathevent))))))))
   (.startfk)
   fk))