Nimbus server, 首先從啟動命令開始, 同樣是使用storm命令"storm nimbus”來啟動
看下源碼, 此處和上面client不同, jvmtype="-server", 最終調用"backtype.storm.daemon.nimbus"的main
nimbus是用clojure實現的, 但是clojure是基于JVM的, 所以在最終發布的時候會產生nimbus.class,
所以在用戶使用的時候完全可以不知道clojure, 看上去所有都是Java. clojure只是用于提高開發效率而已.

1. Nimbus啟動過程

bin/storm

def nimbus(klass="backtype.storm.daemon.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. """ cppaths = [CLUSTER_CONF_DIR] jvmopts = parse_args(confvalue("nimbus.childopts", cppaths)) + [ "-Dlogfile.name=nimbus.log", "-Dlogback.configurationFile=" + STORM_DIR + "/logback/cluster.xml", ] exec_storm_class(klass, jvmtype="-server", extrajars=cppaths, jvmopts=jvmopts)

storm-core/backtype/storm/daemon/nimbus.clj

;; 啟動nimbus的主方法 (defn -main [] ;; main前面加上-, 表示是public的. 所以bin/storm能直接調用nimbus.clj的main方法(-launch (standalone-nimbus))) ;; 同樣launch也是一個public方法. standalone-nimbus是一個方法, clojure對于沒有參數的方法可以省略()(defn -launch [nimbus] ;; launch的參數是一個Nimbus對象, 所以上面standalone-nimbus方法的返回值是Nimbus(launch-server! (read-storm-config) nimbus))

注意在clojure中的函數命名規范,-functionname表示該函數是public的,如上面的-main,調用該函數的時候,不需要加-,使用main即可。
而與此相對的是defn-,這個表示該函數是私有函數,不可在外部調用。

1) standalone-nimbus

nimbus的main, 最終會調到launch-server!, conf參數是調用read-storm-config讀出的配置參數,
而nimbus是INimbus接口(backtype.storm.scheduler.INimbus)的實現, 可以參考standalone-nimbus.
storm-core/backtype/storm/scheduler/INimbus.java

public interface INimbus {void prepare(Map stormConf, String schedulerLocalDir); /**Returns all slots that are available for the next round of scheduling.在下一次調度中可用的槽位 * A slot is available for scheduling 如果槽位是空閑的且可以被分配的, 或者雖然被使用但可以被重新分配的. 都是可以被調度的 * if it is free and can be assigned to, or if it is used and can be reassigned. */ Collection<WorkerSlot> allSlotsAvailableForScheduling(Collection<SupervisorDetails> existingSupervisors, Topologies topologies, Set<String> topologiesMissingAssignments); // this is called after the assignment is changed in ZK void assignSlots(Topologies topologies, Map<String, Collection<WorkerSlot>> newSlotsByTopologyId); // map from node id to supervisor details String getHostName(Map<String, SupervisorDetails> existingSupervisors, String nodeId); IScheduler getForcedScheduler(); } ;; 返回一個實現了INimbus接口的對象. 由于不想創建這種類型, 使用reify匿名對象 (defn standalone-nimbus [] ;; 沒有參數. clojure中[]使用的地方有: let綁定, 方法的參數, vector(reify INimbus ;; reify: 具體化匿名數據類型: 需要一個實現了某一協議/接口的對象，但是不想創建一個命名的數據類型. 匿名類;; 下面的方式都是INimbus接口的實現方法(prepare [this conf local-dir]) ;; this可以看做是一個隱式參數, prepare方法實際只有2個參數的(allSlotsAvailableForScheduling [this supervisors topologies topologies-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))) ))

這里面有好幾個語法點: ->>, mapcat if-let
mapcat, (mapcat f & colls) 和普通map不同的是, 會對map執行的結果執行concat操作等于(concat (map f &colls))
依次對colls中的每個集合運用函數f, 最后將每個結果合并起來. (mapcat f collections)的map不是數據結構意義的映射. 而是一個遍歷操作.
普通的map版本是: (map f collection), 用java來描述就是for(Object o : collection) func(o). 集合中的每個元素會作為函數f的參數.
上面的(mapcat (fn [s] ...))并沒有看到collections. 這個要結合->> supervisors來一起分析.
->> supervisors (mapcat fun) 實際上等價于(mapcat fun supervisors). 由于mapcat的返回值是map,根據接口的定義返回值是一個集合Collection
所以(mapcat)表達式后面的set的意思是將(mapcat)表達式的返回值轉換為set, (mapcat)表達式的返回值會跟在set后面作為最后一個Item.
達到連續調用的功能. ->>和->的區別是->是將返回值作為下一個表達式的第二個Item, 而->>是作為下一個表達式的最后一個Item.

supervisors不是Supervisor列表, 其類型是SupervisorDetails. mapcat后面緊跟的函數的參數類型對應的是collections=supervisors的類型.
WorkerSlot需要兩個參數id和port. 所以這個方法返回的是Collection, 對應接口INimbus的返回類型.

getHostName的參數supervisors和allSlotsAvailableForScheduling的supervisors是一樣的.
通過supervisors.get(node-id)獲取對應的supervisor. 所以我們可以猜測supervisors是一個Map.
storm-core/backtype/storm/scheduler/SupervisorDetails.java

public class SupervisorDetails {String id;String host; // hostname of this supervisor Object meta; Object schedulerMeta; // meta data configured for this supervisor Set<Integer> allPorts; // all the ports of the supervisor }

Nimbus要分配任務給Supervisor上的Worker進行工作, 而每個Supervisor會有多個worker. 配置文件中可以為一個supervisor配置多個slot port.

2) read-storm-config

閱讀源碼其實都會遵循一個范式,那就是程序的入口在哪,配置文件是在什么時候讀入的。那么好,現在就來講配置參數的讀入,在上面的-launch函數中,
已經可以見到用以讀取配置文件的函數了,那就是read-storm-config。非常狗血的是, 在 nimbus.clj 中有一個名稱非常類似的函數稱為read-storm-conf,這個可不是來讀取storm cluster的配置信息,它其實是用來讀取Topology的配置內容的。read-storm-config定義于config.clj中,此時你會說等等,沒見到有地方
import或是use backtype.storm.config啊。這一切都被包裝了,它們統統被放到bootstrap.clj中了。注意到這行沒 (bootstrap)
好了, 上述有關文件引用的疑問解決之后, 還是回到正題, 看看read-storm-config的定義吧。storm默認的配置文件使用的是yaml格式,一定要找到使用yaml parser的地方。
storm-core/backtype/storm/config.clj

(defn read-storm-config [](let [conf (clojurify-structure (Utils/readStormConfig))] ;; let中參數conf被賦值為右側的表達式的值. 和java方法參數不同, let中參數可以被計算 (validate-configs-with-schemas conf) ;; 對conf進行驗證 conf)) ;; 返回這個conf

真正實現對配置文件storm.yaml進行讀取的是由java代碼來實現的,readStormConfig定義于Utils.java中。
storm-core/backtype/storm/utils/Utils.java

public static Map readDefaultConfig() { return findAndReadConfigFile("defaults.yaml", true); } public static Map readStormConfig() { Map ret = readDefaultConfig(); // 首先讀取defaults.yaml的配置 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); // 其次讀取storm.yaml中的配置 ret.putAll(readCommandLineOpts()); // 最后是命令行的參數, 這個優先級更高 return ret; } public static Map findAndReadConfigFile(String name, boolean mustExist) { HashSet<URL> resources = new HashSet<URL>(findResources(name)); URL resource = resources.iterator().next(); Yaml yaml = new Yaml(); Map ret = (Map) yaml.load(new InputStreamReader(resource.openStream())); if(ret==null) ret = new HashMap(); return new HashMap(ret); // 解析storm.yaml文件, 返回HashMap } public static List<URL> findResources(String name) { Enumeration<URL> resources = Thread.currentThread().getContextClassLoader().getResources(name); List<URL> ret = new ArrayList<URL>(); while(resources.hasMoreElements()) { ret.add(resources.nextElement()); } return ret; }

終于看到神秘的Yaml了,那么Yaml這個類又是由誰提供的呢,看看Utils.java的 開頭部分有這么一句話: import org.yaml.snakeyaml.Yaml;
再看看在storm-core/project.clj中定義的dependencies: [org.yaml/snakeyaml "1.11"]
至此,yaml文件的解析及其依賴關系的解決探索完畢。在新版本的storm中使用了maven管理. 可以查看pom.xml

3) storm.yaml

conf/storm.yaml

# storm.zookeeper.servers: # - "server1" # - "server2" # # nimbus.host: "nimbus" # # ##### These may optionally be filled in: # ## List of custom serializations # topology.kryo.register: # - org.mycompany.MyType # - org.mycompany.MyType2: org.mycompany.MyType2Serializer # ## List of custom kryo decorators # topology.kryo.decorators: # - org.mycompany.MyDecorator # ## Locations of the drpc servers # drpc.servers: # - "server1" # - "server2" ## Metrics Consumers # topology.metrics.consumer.register: # - class: "backtype.storm.metrics.LoggingMetricsConsumer" # parallelism.hint: 1 # - class: "org.mycompany.MyMetricsConsumer" # parallelism.hint: 1 # argument: # - endpoint: "metrics-collector.mycompany.org" storm.zookeeper.servers: - 127.0.0.1 storm.zookeeper.port: 2181 nimbus.host: "127.0.0.1" storm.local.dir: "/home/hadoop/data/storm" supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703

在配置文件中需要至少回答以下三個問題
1. zookeeper server在哪臺機器上運行,具體就來說就是ip地址啦
2. nimbus在哪運行,可以填寫ip地址或域名
3. 在每臺supervisor運行的機器上可以啟幾個slot,指定這些slot監聽的端 口號
2. thrift RPC
1) thrift
網絡結點之間的消息交互一般會牽涉到兩個基本的問題,
? 消息通道的建立
? 消息的編解碼
如果每變化一個需求就手工來重寫一次,一是繁瑣,二是易錯。為了一勞永逸的解決此類問題,神一樣的工具就出現了,如google protolbuffer,如thrift.
thrift的使用步驟如下

編寫后綴名為thrift的文件,使用工具生成對應語言的源碼,thrift支持的語言很多的,什么c,c++,java,python等,統統不是問題。
實現thrift client
實現thrift server
thrift server需要實現thrift文件中定義的service接口。更為具體的信息可以通過閱讀官方文檔來獲得。這里有個thrift java的示例.

(1). 編寫thrift文件：add.thrift

namespace java com.zqh.code.thrift.server // defines the namespace typedef i32 int // typedefs to get convenient names for your types service AdditionService { // defines the service to add two numbers int add(1:int n1, 2:int n2), // defines a method }

(2). 編譯：thrift --gen java add.thrift 會在當前目錄生成gen-java/$namespace$/AdditionService
(3). Service：Interface的實現類

public class AdditionServiceHandler implements AdditionService.Iface { public int add(int n1, int n2) throws TException { return n1 + n2; } }

實現類具體實現了thrift文件定義的接口方法.
(4). Server

public class MyServer {public static void start(AdditionService.Processor<AdditionServiceHandler> processor) { TServerTransport serverTransport = new TServerSocket(9090); // 服務端Socket TServer server = new TSimpleServer(new Args(serverTransport).processor(processor)); System.out.println("Starting the simple server..."); server.serve(); } public static void main(String[] args) { start(new AdditionService.Processor<AdditionServiceHandler>(new AdditionServiceHandler())); } }

服務端通過TServerSocket暴露出服務端口, 客戶端要通過這個端口連接.
實現類Handler的實例要作為生成的AdditionService.Processor的參數.
Args需要TServerTransport作為參數, 然后調用processor方法, 該方法需要AdditionServiceProcessor參數.
這個過程類似于將自定義實現類Handler注冊到服務端上. 接著啟動服務器.
(5). Client

public class AdditionClient {public static void main(String[] args) { TTransport transport = new TSocket("localhost", 9090); transport.open(); TProtocol protocol = new TBinaryProtocol(transport); AdditionService.Client client = new AdditionService.Client(protocol); System.out.println(client.add(100, 200)); transport.close(); } }

客戶端要建立到服務端的連接, 需要提供Server的host和port. 根據TTransport構造出和服務端進行通訊的一個協議.
這個協議傳給自動生成的AdditionService的Client內部類, 會生成一個類似服務端的代理對象.
接著就可以使用這個代理對象調用thrift協議提供的方法.

分布式測試: 可以在兩臺機器上測試. 第一二步都需要在兩臺機器上操作: 編寫thrift文件, 編譯.
然后在第一臺機器操作3: 自定義實現類; 4: Server. 在第二臺機器上操作5: Client. 最后分別運行兩臺機器的Server和Client.

2) nimbus thrift server

有了thrift這個背景,我們再重新拾起上述的代碼執行路徑。上頭講到程序執行至

(defn -launch [nimbus] ;; launch的參數是一個Nimbus對象, 所以上面standalone-nimbus方法的返回值是Nimbus(launch-server! (read-storm-config) nimbus))(defn launch-server! [conf nimbus] ;; 讓nimbus作為一個thrift server運行起來(validate-distributed-mode! conf) ;; 分布式模式下才會啟動thrift server(let [service-handler (service-handler conf nimbus) ;; 自定義實現類, 實現storm.thrift中service Nimbus定義的接口方法options (-> (TNonblockingServerSocket. (int (conf NIMBUS-THRIFT-PORT))) ;; 服務端的ServerSocket(THsHaServer$Args.) ;; TServerSocket作為TServer.Args內部類的參數. 創建了Args args對象 ->表示插入第二個位置(.workerThreads 64) ;; 上面new Args(TServerSocket)會作為這里的第二個位置, 即args.workerThreads(64) (.protocolFactory (TBinaryProtocol$Factory. false true (conf NIMBUS-THRIFT-MAX-BUFFER-SIZE))) (.processor (Nimbus$Processor. service-handler)) ;; args作為這里的第二個位置,即調用了args.processor ;; new Nimbus.Processor(service-handler), 自定義實現類作為Nimbus.Processor的參數, ;; processor會作為參數再傳給args.processor() ) ;; 最終返回的是TServer.AbstractServerArgs, 會作為TServer構造函數的參數 server (THsHaServer. (do (set! (. options maxReadBufferBytes)(conf NIMBUS-THRIFT-MAX-BUFFER-SIZE)) options))] (.addShutdownHook (Runtime/getRuntime) (Thread. (fn [] (.shutdown service-handler) (.stop server)))) (log-message "Starting Nimbus server...") ;; 上面添加了一個關閉鉤子. 類似回調函數. 當關閉Nimbus的thrift服務時, 會觸發這個函數執行 (.serve server))) ;; 啟動TServer, 即啟動Nimbus的thrift服務

launch-server!說白了,就是讓nimbus作為一個thrift server運行起來, 那么storm.thrift中service指定的各個接口函數實現在service-handler中完成。
對比clojure版本的創建thrift server的過程, 其實和上面java示例是一樣的, 只不過換了不同的實現類. 以下是java-clojure的代碼對比.
new AdditionServiceHandler() (service-handler conf nimbus)
new AdditionService.Processor(new AdditionServiceHandler()) (Nimbus$Processor. service-handler)
TServerTransport serverTransport = new TServerSocket(9090); (TNonblockingServerSocket. (int (conf NIMBUS-THRIFT-PORT)))
new Args(serverTransport) -> (TNonblockingServerSocket...) (THsHaServer$Args.)
new Args(serverTransport).processor(processor) -> (TNonblockingServerSocket...) (THsHaServer$Args.) (.processor (Nimbus$Processor. ..))
TServer server = new TSimpleServer(new Args(serverTransport).processor(processor)); server (THsHaServer… options)
server.serve(); (.serve server)

service-handler可是一個大家伙。對比一下 service-handler可以發現,在storm.thrift中的定義的Nimbus服務,
其接口在 service-handler中一一得以實現。 以下是storm.thrift中關于service Nimbus的聲明。
storm-core/storm.thrift

namespace java backtype.storm.generated service Nimbus {void submitTopology(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology) throws (1: AlreadyAliveException e, 2: InvalidTopologyException ite); void submitTopologyWithOpts(1: string name, 2: string uploadedJarLocation, 3: string jsonConf, 4: StormTopology topology, 5: SubmitOptions options) throws ...; void killTopology(1: string name) throws (1: NotAliveException e); void killTopologyWithOpts(1: string name, 2: KillOptions options) throws (1: NotAliveException e); void activate(1: string name) throws (1: NotAliveException e); void deactivate(1: string name) throws (1: NotAliveException e); void rebalance(1: string name, 2: RebalanceOptions options) throws (1: NotAliveException e, 2: InvalidTopologyException ite); // need to add functions for asking about status of storms, what nodes they're running on, looking at task logs string beginFileUpload(); void uploadChunk(1: string location, 2: binary chunk); void finishFileUpload(1: string location); string beginFileDownload(1: string file); binary downloadChunk(1: string id); //can stop downloading chunks when receive 0-length byte array back string getNimbusConf(); // returns json ClusterSummary getClusterInfo(); // stats functions TopologyInfo getTopologyInfo(1: string id) throws (1: NotAliveException e); string getTopologyConf(1: string id) throws (1: NotAliveException e); //returns json StormTopology getTopology(1: string id) throws (1: NotAliveException e); StormTopology getUserTopology(1: string id) throws (1: NotAliveException e); }

這個文件還指定了其他一些struct結構的數據類型, 比如StormTopology, TopologySummary, ClusterSummary, TopologyInfo等.
編譯storm.thrift文件生成的代碼在namespace指定的位置: backtype.storm.generated
storm-core/genthrift.sh

thrift7 --gen java:beans,hashcode,nocamel --gen py:utf8strings storm.thrift

現在來回顧下storm的thrift RPC的整體流程.
1. 編寫 storm.thrift
2. 編譯 genthrift.sh, 會在backtype.storm.generated生成Nimbus.java接口類. 其中含有內部類Iface(Service), Processor(Server), Client(Client)
3. Service服務類: nimbus.clj中的service-handler方法的返回值. 其應該實現Nimbus.Iface接口. 所以service-handler使用reify Nimbus$Iface
4. Server服務端: launch-server!中創建thrift的TServer, 并啟動. 使用了Nimbus.Processor, 傳入service-handler自定義服務實現類
5. Client客戶端: StormSubmitter中localNimbus!=null時, 使用NimbusClient即Nimbus.Client調用RPC定義的接口方法

注意: 對于本地模式, 在StormSubmitter中直接使用Nimbus.Iface localNimbus對象. 這個對象的實現類應該就是service-handler.
對于分布式模式, StormSubmitter作為客戶端, 會通過client調用RPC定義的接口方法. 即storm.thrift中定義的方法. 所以service-handler要實現這些方法!

• 2015年04月14日發布

轉載于:https://www.cnblogs.com/catkins/p/5252480.html

總結

以上是生活随笔為你收集整理的Nimbus三Storm源码分析--Nimbus启动过程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。