def nimbus():
    """Syntax: [storm nimbus]

Launches the nimbus daemon. This command should be run under
supervision with a tool like daemontools or monit.

See Setting up a Storm cluster for more information.
(https://github.com/nathanmarz/storm/wiki/Setting-up-a-Storm-cluster)
"""
    cppaths = [STORM_DIR + "/log4j", STORM_DIR + "/conf"]
    childopts = confvalue("nimbus.childopts", cppaths) + " -Dlogfile.name=nimbus.log -Dlog4j.configuration=storm.log.properties"
    exec_storm_class(
        "backtype.storm.daemon.nimbus",
        jvmtype="-server",
        extrajars=cppaths,
        childopts=childopts)

launch-server!

来看看nimbus的main, 最终会调到launch-server!, conf参数是调用read-storm-config读出的配置参数, 

而nimbus是INimbus接口(backtype.storm.scheduler.INimbus)的实现, 可以参考standalone-nimbus

(defn -main []
  (-launch (standalone-nimbus)))

(defn -launch [nimbus]
  (launch-server! (read-storm-config) nimbus))

(defn launch-server! [conf nimbus]
  (validate-distributed-mode! conf)
  (let [service-handler (service-handler conf nimbus)
        options (-> (TNonblockingServerSocket. (int (conf NIMBUS-THRIFT-PORT)))
                    (THsHaServer$Args.)
                    (.workerThreads 64)
                    (.protocolFactory (TBinaryProtocol$Factory.))
                    (.processor (Nimbus$Processor. service-handler))
                    )
       server (THsHaServer. options)]
    (.addShutdownHook (Runtime/getRuntime) (Thread. (fn [] (.shutdown service-handler) (.stop server))))
    (log-message "Starting Nimbus server...")
    (.serve server)))

1. service-handler

(defserverfn service-handler [conf inimbus]
    (reify Nimbus$Iface
      ...)
)

(defmacro defserverfn [name & body]
  `(let [exec-fn# (fn ~@body)]
    (defn ~name [& args#]0
      (try-cause
        (apply exec-fn# args#)
      (catch InterruptedException e#
        (throw e#))
      (catch Throwable t#
        (log-error t# "Error on initialization of server " ~(str name))
        (halt-process! 13 "Error on initialization")
        )))))

let [nimbus (nimbus-data conf inimbus)]

(defn nimbus-data [conf inimbus]
  (let [forced-scheduler (.getForcedScheduler inimbus)]
    {:conf conf
     :inimbus inimbus
     :submitted-count (atom 0) ;记录多少topology被submit
     :storm-cluster-state (cluster/mk-storm-cluster-state conf) ;抽象Zookeeper接口(Zookeeper用于存放cluster state)
     :submit-lock (Object.) ;创建锁对象,用于各个topology之间的互斥操作, 比如建目录
     :heartbeats-cache (atom {}) ;记录各个Topology的heartbeats的cache
     :downloaders (file-cache-map conf)
     :uploaders (file-cache-map conf)
     :uptime (uptime-computer)
     :validator (new-instance (conf NIMBUS-TOPOLOGY-VALIDATOR))
     :timer (mk-timer :kill-fn (fn [t]
                                 (log-error t "Error when processing event")
                                 (halt-process! 20 "Error when processing an event")
                                 ))
     :scheduler (mk-scheduler conf inimbus)
     }))

(system-topology! total-storm-conf topology) ;; this validates the structure of the topology
(defn system-topology! [storm-conf ^StormTopology topology]
  (validate-basic! topology)
  (let [ret (.deepCopy topology)]
    (add-acker! storm-conf ret)
    (add-metric-components! storm-conf ret)    
    (add-system-components! storm-conf ret)
    (add-metric-streams! ret)
    (add-system-streams! ret)
    (validate-structure! ret)
    ret
    ))

(setup-storm-code conf storm-id uploadedJarLocation storm-conf topology)

借用这张图, 比较清晰, 先创建目录, 并将Jar move到当前目录

再将topology对象和conf对象都序列化保存到目录中

(.setup-heartbeats! storm-cluster-state storm-id)

(setup-heartbeats! [this storm-id]
  (mkdirs cluster-state (workerbeat-storm-root storm-id)))

(defn mkdirs [^CuratorFramework zk ^String path]
  (let [path (normalize-path path)]
    (when-not (or (= path "/") (exists-node? zk path false))
      (mkdirs zk (parent-path path))
      (try-cause
        (create-node zk path (barr 7) :persistent)
        (catch KeeperException$NodeExistsException e
          ;; this can happen when multiple clients doing mkdir at same time
          ))
      )))

(activate-storm! [this storm-id storm-base]
  (set-data cluster-state (storm-path storm-id) (Utils/serialize storm-base))
  )

(defn storm-path [id]
  (str STORMS-SUBTREE "/" id)) ;/storms/id

(defn set-data [^CuratorFramework zk ^String path ^bytes data]
  (.. zk (setData) (forPath (normalize-path path) data)))