JStorm系列：概念與編程模型

1、集群架構

JStorm從設計的角度，就是一個典型的調度系統，簡單集群的架構如下圖所示，其中Nimbus可增加一個備節點，多個Supervisor節點組成任務執行集群。

簡單集群示意圖

1.1、Nimbus

Nimbus是作為整個集群的調度器角色，負責分發topology代碼、分配任務，監控集群運行狀態等，其主要通過ZK與supervisor交互。可以和Supervisor運行在同一物理機上，JStorm中Nimbus可採用主從備份，支持熱切。

1.2、Supervisor

Supervisor 是集群中任務的執行者，負責運行具體任務以及關閉任務。其從ZK中監聽nimbus的指令，然後接收分發代碼和任務並執行、監控反饋任務執行情況。

1.3 、Zookeeper

ZK是整個系統中的協調者，Nimbus的任務調度通過ZK下發至Supervisor來執行。

2、Topology編程模型

Topology是一個可以在JStorm中運行的任務的抽象表達，在JStorm的topology中，有兩種組件：spout和bolt。下面是一張比較經典的Topology結構圖。每一個topology，既可以有多個spout，代表同時從多個數據源接收消息，也可以多個bolt，來執行不同的業務邏輯。一個topology會一直運行直到你手動kill掉，JStorm自動重新分配執行失敗的任務。

在JStorm中有對於流stream的抽象，流是一個不間斷的無界的連續tuple，注意JStorm在建模事件流時，把流中的事件抽象為tuple即元組。

我們可以認為spout就是一個一個的水龍頭，並且每個水龍頭裡流出的水是不同的tuple，我們想拿到哪種水tuple就擰開哪個水龍頭，然後使用管道將水龍頭的水tuple導向到一個水處理器(bolt)，水處理器bolt處理後再使用管道導向另一個處理器或者存入容器中。

Topology示意圖

JStorm將上圖抽象為Topology即拓撲，拓撲結構是有向無環的，拓撲是Jstorm中最高層次的一個抽象概念，它可以被提交到Jstorm集群執行，一個拓撲就是一個數據流轉換圖，圖中每個節點是一個spout或者bolt，圖中的邊表示bolt訂閱了哪些流，當spout或者bolt發送元組到流時，它就發送元組到每個訂閱了該流的bolt。

2.1、spout

JStorm認為每個stream都有一個stream源，也就是原始元組的源頭，所以它將這個源頭抽象為spout，spout可能是連接消息中間件(如MetaQ， Kafka， TBNotify等)，並不斷發出消息，也可能是從某個隊列中不斷讀取隊列元素並裝配為tuple發射。

JStorm框架對spout組件定義了一個主要方法：nextTuple，顧名思義，就是獲取下一條消息。執行時，可以理解成JStorm框架會不停地調這個介面，以從數據源拉取數據並往bolt發送數據。

Tuple是一次消息傳遞的基本單元，tuple里的每個欄位一個名字,並且不同tuple的對應欄位的類型必須一樣。tuple的欄位類型可以是： integer, long, short, byte, string, double, float, boolean和byte array。還可以自定義類型，只要實現對應的序列化器。

JStorm中與spout相關的介面主要是ISpout和IRichSpout、IBatchSpout，後兩介面實現了對ISpout介面的上層封裝。

ISpout介面主要方法：

open：在worker中初始化該ISpout時調用，一般用來設置一些屬性：比如從spring容器中獲取對應的Bean。

close：和open相對應(在要關閉的時候調用)。

activate：從非活動狀態變為活動狀態時調用。

deactivate：和activate相對應(從活動狀態變為非活動狀態時調用)。

nextTuple：JStorm希望在每次調用該方法的時候，它會通過collector.emit發射一個tuple。

ack：jstorm發現msgId對應的tuple被成功地完整消費會調用該方法。

fail：和ack相對應(jstorm發現某個tuple在某個環節失敗了)。和ack一起保證tuple一定被處理。

2.2、bolt

JStorm將tuple的中間處理過程抽象為Bolt，bolt可以消費任意數量的輸入流，只要將流方嚮導向該bolt，同時它也可以發送新的流給其他bolt使用，這樣一來，只要打開特定的spout(管口)再將spout中流出的tuple導向特定的bolt，然後bolt對導入的流做處理後再導向其他bolt或者目的地。

bolt代表處理邏輯，bolt收到消息之後，對消息做處理(即執行用戶的業務邏輯)，處理完以後，既可以將處理後的消息繼續發送到下游的bolt，這樣會形成一個處理流水線(不過更複雜的情況應該是個有向圖);也可以直接結束。

bolt組件主要方法：execute，這個介面就是用戶用來處理業務邏輯的地方。

通常一個流水線的最後一個bolt，會做一些數據的存儲工作，比如將實時計算出來的數據寫入DB、HBase等，以供前台業務進行查詢和展現。Bolts可以發射多條消息流， 使用OutputFieldsDeclarer.declareStream定義stream，使用OutputCollector.emit來選擇要發射的stream。

在保證不丟消息的場景中，在bolts必須要為它處理的每一個tuple調用OutputCollector的ack方法，以通知JStorm這個tuple被處理完成了，從而通知這個tuple的發射者spouts。 一般的流程是： bolts處理一個輸入tuple, 發射0個或者多個tuple, 然後調用ack通知JStorm自己已經處理過這個tuple了。JStorm提供了一個IBasicBolt會自動調用ack。

JStorm中與Bolt相關的介面主要是IBolt，IRichBolt，IBasicBolt和IBatchBolt，後面介面實現了對IBolt介面的上層封裝。

IBolt介面的主要方法：

prepare：在worker中初始化該IBolt時調用，一般用來設置一些屬性：比如從spring容器中獲取對應的Bean。

cleanup：和prepare相對應(在顯示關閉topology的時候調用)

execute：處理jstorm發送過來的tuple。

2.3、Tuple

JStorm將流中數據抽象為tuple，一個tuple就是一個值列表value list，list中的每個value都有一個name，tuple可以由任意類型組合而成，因為storm是分散式的，所以它需要知道在task間如何序列化和反序列化數據的。storm使用Kryo進行序列化，Kryo是java開發中一個快速靈活序列器。默認情況下，storm可以序列化基礎類型，比如字元串，位元組，數組，ArrayList, HashMap, HashSet和 Clojure 集合類型，如果需要使用其他類型，需要自定義序列器。拓撲的每個節點都要說明它所發射出的元組的欄位的name，其他節點只需要訂閱該name就可以接收處理。

在spout和Bolt組件中，使用declareOutputFields方法定義發射出的tuple的欄位名。

3、小結

本文主要講述了JStorm中集群的架構以及Topology編程模型方面的概念知識，後續會更深入的寫一些實踐、運維、原理等方面的文章。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！