storm-0.8.2源码分析之nimbus启动
nimbus启动脚本启动逻辑

通过bin/stormnimbus会启动nimbus进程，类似hadoop的jobtracker。

bin/storm是一个python写的脚本，支持jar,kill,nimbus,supervisor,ui,drpc等等命令。

python的入口函数如下

def main():

if len(sys.argv) <= 1:

print_usage()

sys.exit(-1)

global CONFIG_OPTS

config_list, args = parse_config_opts(sys.argv[1:])

parse_config(config_list)

COMMAND = args[0]

ARGS = args[1:]

(COMMANDS.get(COMMAND, unknown_command))(*ARGS)

最后这条语句将根据不同的命令调用其对应的方法，启动nimbus这里调用是

nimbus(klass="backtype.storm.daemon.nimbus")，在nimbus方法内部会调用exec_storm_class(所有命令所调用的方法内部都会调用这个)，根据指定的入口类backtype.storm.daemon.nimbus去启动程序。exec_storm_class内部其实就是通过exec执行java 参数 包名.类名 来启动对应的程序。

nimbus的clojure入口

nimbus从脚本中执行的时候，会根据之前传递的backtype.storm.daemon.nimbus这个类，去定位到backtype/storm/daemon/nimbus.clj这个clojure的文件（关于clojure的命名空间与实际文件的映射其实和java中是一样的的），其中-main是nimbus.clj的入口函数，相当c++/java中的main入口。

-main非常简单，如下所示

(defn -main []

(-launch (standalone-nimbus)))

下面将依次分析，standalong-numbus，以及-launch args各自做了些什么。

standalong-numbus是个函数，在这个函数内部是实现INimbus的接口，大致过程是用到了clojure.core.protocols中的reify来定义了匿名的类型实现了java中的package backtype.storm.scheduler这个包中的INimbus这个接口(clojure中接口的对应物称为协议)

下面代码中参数列表中的第一个参数this，是类似c++/java中的类的成员函数的第一个隐藏参数this，也就是通过this确定到底是使用这个协议的哪个实现。

先看standalone-nimbus的代码如下

(defn standalone-nimbus []

(reify INimbus

(prepare [this conf local-dir]

)

(allSlotsAvailableForScheduling [this supervisors topologiestopologies-missing-assignments]

(->> supervisors

(mapcat (fn [^SupervisorDetails s]

(for [p (.getMeta s)]

(WorkerSlot. (.getId s)p))))

set ))

(assignSlots [this topology slots]

)

(getForcedScheduler [this]

nil )

(getHostName [this supervisors node-id]

(if-let [^SupervisorDetails supervisor (get supervisors node-id)]

(.getHost supervisor)))

))

allSlotsAvailableForScheduling 按clojure的语义进行转下，代码如下

(set ((mapcat (fn [^SupervisorDetails s]

(for [p (.getMeta s)]

(WorkerSlot. (.getId s)p)))) supervisors)

)

也就是说通过遍历Collection<SupervisorDetails> 类型的变量supervisors取id和meta，用来初始化WorkerSolt对象，mapcat将WorkerSolt形成集合，之后将这个集合作为参数传递给set，使之返回符合接口要求的Collection<WorkerSlot>

getHostName 是取主机名的接口，supervisors 的类型是Map<String,SupervisorDetails> ，node-id的类型是String。其过程是从supervisors中通过node-id去找符合条件的SupervisorDetails对象，如果找到了，就调用SupervisorDetails的getHost方法取得主机名，否者返回的是nil

(getHostName [this supervisors node-id]

(if-let [^SupervisorDetails supervisor (get supervisors node-id)]

(.getHost supervisor)))

)

其他几个接口standalone-nimbus中没有去实现。

注：这里涉及到了clojure与java backtype.storm.scheduler包中几个类的交互。

-launch 将之前的standalone-nimbus做为参数传递给-launch，其代码如下:

(defn -launch [nimbus]

(launch-server! (read-storm-config) nimbus))

配置文件读取

首先，先介绍下(read-storm-config)的引用的过程，nimbus.clj的头部定义命名空间的时候，有一句(:use [backtype.storm.daemon common])引入了common.clj。而common.clj中在定义自身名字空间的时候也引用了其他的其他名字空间(:use [backtype.storm log config util]) ，read-storm-config函数就是在config.clj中实现的，其定义如下：

(defn read-storm-config []

(clojurify-structure (Utils/readStormConfig)))

其中Utils/readStormConfig表示使用了java的backtype.storm.utils包中的Utils类的readStormConfig方法，其定义如下:

publicstatic Map readStormConfig() {

Map ret = readDefaultConfig();

String confFile = System.getProperty("storm.conf.file");

Map storm;

if (confFile==null || confFile.equals("")) {

storm = findAndReadConfigFile("storm.yaml", false);

} else {

storm = findAndReadConfigFile(confFile, true);

}

ret.putAll(storm);

ret.putAll(readCommandLineOpts());

return ret;

}

这个方法就是真正去解析配置的入口，解析过程：先读取默认的defualts.yaml的配置，对于源码来说该文件是在conf目录下，对于release版本则是该文件打到了storm-0.8.2.jar内。

其次，再解析用户配置的storm.yaml中的配置项，如果strom.yaml中有配置项与默认配置文件的配置项有冲突，则会覆盖掉默认配置项。最后，取系统环境变量中设置的storm.options的值，这一般都是没有的，因此这步可以跳过。

注:storm的配置文件用到了yaml这种配置格式，可参考其官方http://www.yaml.org/

再回过头看看(Utils/readStormConfig)取得的配置的map做为参数传递给clojurify-structure做了些什么。

(defn clojurify-structure [s]

(prewalk (fn [x]

(cond (instance? Map x) (into {}x)

(instance? List x) (vec x)

true x))

s))

这个函数的功能是将外部数据结构转换成clojure内部的数据结构，在这里，传进来的是java的map，那么会将其转换成clojure的map。另外，prewalk ([f form])的作用简单来说从集合中取值，该值之后会由经过函数计算后的值来替换掉。

因此, (launch-server!(read-storm-config) nimbus)) 中的(read-storm-config) 就是读取storm的配置文件并返回clojure中的map结构。

启动server

启动server的代码的调度是封装在launch-server这个函数内，先看这个函数

(defn launch-server! [conf nimbus]

(validate-distributed-mode! conf)

(let [service-handler (service-handler conf nimbus)

options (-> (TNonblockingServerSocket. (int (confNIMBUS-THRIFT-PORT)))

(THsHaServer$Args.)

(.workerThreads 64)

(.protocolFactory(TBinaryProtocol$Factory.))

(.processor(Nimbus$Processor. service-handler))

)

server (THsHaServer. options)]

(.addShutdownHook (Runtime/getRuntime) (Thread. (fn [] (.shutdownservice-handler) (.stop server))))

(log-message "Starting Nimbus server...")

(.serve server)))

先看看第一条语句(validate-distributed-mode!conf)的作用是从底层从配置中读取storm.cluster.mode的值，判断storm.cluster.mode的值配置的是否合法，即，是"local"返回true，是"distributed"返回false，如果是其他的话则抛出异常。之后，validate-distributed-mode!内部判断如果storm.cluster.mode的值是"local"则抛出异常。总结起来简单的说，这个函数是判断在defaults.yaml中的配置项storm.cluster.mode的值是否是"distributed"，如果不是的话则会抛出异常(因为，defaults.yaml默认配置的是"distributed")。

;common.clj

(defn validate-distributed-mode! [conf]

(if(local-mode? conf) ;如果是"local"模式会抛出异常

(throw

(IllegalArgumentException. "Cannot start server in localmode!"))))

;config.clj

(defn local-mode? [conf]

(let [mode (conf STORM-CLUSTER-MODE)]; 从配置的map中取storm.cluster.mode的值

(condp = mode

"local" true

"distributed" false

(throw (IllegalArgumentException. ;不是上面两种，抛出异常

(str "Illegal cluster modein conf: " mode)))

)))

看看launch-server!函数中很重要的一句，service-handler(service-handler conf nimbus)

在启动流程的分析过程中，对于这条语句会先简单分析，之后单独拿出来再做详细分析。

(service-handler conf nimbus) 的service-handler其实是一个函数，这个函数不是直接定义的，而是通过defserverfn来定义的，可以在nimbus.clj中找到

(defserverfn service-handler [conf inimbus]...) 其中...是body太长，这里就省略了。

body内部的主要功能是使用clojure的关键字reify去匿名实现一些接口。

内部实现的最主要的接口是package backtype.storm.generated包中的Nimbus类内部的Iface接口。另外还实现了两个接口，分别是package backtype.storm.daemon中的Shutdownable接口，和common.clj中申明的DaemonCommon协议。

这里需要引申说一下，storm是使用thrift作为网络通讯中间件，它提供了序列化和RPC功能，还支持多语言通讯。因此storm是支持多语言客户端的。

使用thrift工具，对编写的.thrift文件生成指定语言的代码，之后只需要实现.thrift中生成的service对应的接口的代码，组织个server启动下，就完成了RPC服务端的功能了。

在storm中.thrift文件位于src/storm.thrift，使用thrift工具生成的代码位于src\jvm\backtype\storm\generated，也就是都属于jvm目录下的packagebacktype.storm.generated。src/storm.thrift内有一个service Nimbus接口，因此在生成backtype.storm.generated包中的Nimbus类内部有接口Iface需要实现，这个就是RPC接口，server实现这些接口，客户端就可以直接调用这些接口，可以像本地调用函数一样调用。

再看分析下(defserverfn service-handler [conf inimbus] ...) 的功能，这条语句相当于是通过宏包裹了函数，并对原有函数执行体进行异常捕捉。可以看到defserverfn的宏定义如下所示：

;common.clj中的宏defserverfn的实现如下所示

(defmacro defserverfn [name & body]

`(let [exec-fn# (fn ~@body)]

(defn ~name [& args#]

(try-cause

(apply exec-fn# args#)

(catch InterruptedException e#

(throw e#))

(catch Throwable t#

(log-error t# "Error on initialization of server " ~(strname))

(halt-process! 13 "Error on initialization")

)))))

;以service-handler带人到宏内,简化后相当于是

;defn server-handler [& args]

;applay (fn [conf inimbus] ... ) [&args]

;其中,server-handler是函数名,而& args就是 confnimbus，三个点号就是原有的函数执行体。

回到launch-server中，看下options最后存的是什么。

options (-> (TNonblockingServerSocket.(int (conf NIMBUS-THRIFT-PORT)))

(THsHaServer$Args.)

(.workerThreads 64)

(.protocolFactory(TBinaryProtocol$Factory.))

(.processor(Nimbus$Processor. service-handler))

)

先解释下->的用法：

在clojure中->的作用是将参数一次传递给下一个form，作为这个form的第一个参数。如果存在多个form这将第一个form计算的结果传递个下一个form，作为下一个form的第一个参数。

因此options后面的代码真正展开就就是如下调用形式：

(.processor

(.protocolFactory

(.workerThreads

(THsHaServer$Args.

(TNonblockingServerSocket. (int(conf NIMBUS-THRIFT-PORT))))

64)

(TBinaryProtocol$Factory.))

(Nimbus$Processor. service-handler))

这里是涉及到了thriftjava库中的相关代码的调用，因此将引入thritf的代码方便解释流程。首先创建一个packageorg.apache.thrift.transport中的TNonblockingServerSocket对象，将给对象作为参数传递给package org.apache.thrift.server中的THsHaServer的内部类Args的构造函数，这样就得到一个Args对象。对Args对象调用workerThreads 方法，设置woker线程数为64，并返回原来的Args对象。然后调用Args的基类AbstractServerArgs(位于TServer.java中)中的protocolFactory方法，并创建一个package org.apache.thrift.protocol中的TBinaryProtocol的内部类Factory的对象，将Factory对象作为AbstractServerArgs.protocolFactory方法的参数，内部是设置inputProtocolFactory，outputProtocolFactory为传递进去的factory，然后返回this,也就是返回Args对象。

最后，调用Args的基类AbstractServerArgs的processor方法，内部就是传递进去的processor参数(由Nimbus$Processor. service-handler这条语句生成)，生成processor的工厂类的对象，之后返回this,也就是Args对象。

因此options就是保存了Args对象。

再来看看let内的最后一条语句 server(THsHaServer. options)。其中(THsHaServer. options)是调用了packageorg.apache.thrift.server中的THsHaServer的构造函数来生成一个THsHaServer对象，构造函数是需要一个Args对象，而options保存的就是Args对象。因此server保存了一个THsHaServer对象。

let块的表达式的第一条语句如下：

(.addShutdownHook (Runtime/getRuntime)(Thread. (fn [] (.shutdown service-handler) (.stop server))))

这句的调用转为成java其实就是Runtime.getRuntime.addShutdownHook(Thread hook)，而参数hook，就是上面的(Thread.(fn [] (.shutdown service-handler) (.stop server)))

这样的话，就容易理解了，它的作用是，增加JVM停止时要处理的事件，当JVM停止时会调用这些注册的hook。

在看看let块的表达式的最后一条语句(.serveserver) ，这就是启动thrift server。serve方法是在THsHaServer的基类AbstractNonblockingServer中，serve方法的功能就是接收连接处理调用。

其实，熟悉java中如何启动一个thriftserver，对于上诉的在clojure启动thrift server也会比较清楚。如果不清楚在java中是如何启动thriftserver的，可以自行去参考thrift中自带的例子。