問題導讀：
1.hadoop有master與slave，Storm與之對應的節點是什么？
2.Storm控制節點上面運行一個后台程序被稱之為什么？
3.Supervisor的作用是什么？
4.Topology與Worker之間的關系是什么？
5.Nimbus和Supervisor之間的所有協調工作有master來完成，還是Zookeeper集群完成？
6.storm穩定的原因是什么？
7.如何運行Topology？
strom jar all-your-code.jar backtype.storm.MyTopology arg1 arg2
8.spout是什么？
9.bolt是什么？
10.Topology由兩部分組成？
11.stream grouping有幾種？
伴隨着信息科技日新月異的發展，信息呈現出爆發式的膨脹，人們獲取信息的途徑也更加多樣、更加便捷，同時對於信息的時效性要求也越來越高。舉個搜索場景中的例子，當一個賣家發布了一條寶貝信息時，他希望的當然是這個寶貝馬上就可以被賣家搜索出來、點擊、購買啦，相反，如果這個寶貝要等到第二天或者更久才可以被搜出來，估計這個大哥就要不滿意了。再舉一個推薦的例子，如果用戶昨天在淘寶上買了一雙襪子，今天想買一副泳鏡去游泳，但是卻發現系統在不遺余力地給他推薦襪子、鞋子，根本對他今天尋找泳鏡的行為視而不見，估計這哥們心里就會不痛快了呀。其實稍微了解點背景知識的碼農們都知道，這是因為后台系統做的是每天一次的全量處理，而且大多是在夜深人靜之時做的，那么你今天白天做的事情當然要明天才能反映出來啦。
一、一個實時計算系統要解決的問題
全量數據處理使用的大多是鼎鼎大名的hadoop或者hive，作為一個批處理系統，hadoop以其吞吐量大、自動容錯等優點，在海量數據處理上得到了廣泛的使用。但是，hadoop不擅長實時計算，因為它天然就是為批處理而生的，這也是業界一致的共識。否則最近這兩年也不會有s4,storm,puma這些實時計算系統如雨后春筍般冒出來啦。先拋開s4,storm,puma這些系統不談，我們首先來看一下，如果讓我們自己設計一個實時計算系統，我們要解決哪些問題。
- 低延遲。都說了是實時計算系統了，延遲是一定要低的。
- 高性能。性能不高就是浪費機器，浪費機器是要受批評的哦。
- 分布式。系統都是為應用場景而生的，如果你的應用場景、你的數據和計算單機就能搞定，那么不用考慮這些復雜的問題了。我們所說的是單機搞不定的情況。
- 可擴展。伴隨着業務的發展，我們的數據量、計算量可能會越來越大，所以希望這個系統是可擴展的。
- 容錯。這是分布式系統中通用問題。一個節點掛了不能影響我的應用。
好，如果僅僅需要解決這5個問題，可能會有無數種方案，而且各有千秋，隨便舉一種方案，使用消息隊列+分布在各個機器上的工作進程就ok啦。我們再繼續往下看。
- 容易在上面開發應用程序。親，你設計的系統需要應用程序開發人員考慮各個處理組件的分布、消息的傳遞嗎？如果是，那有點麻煩啊，開發人員可能會用不好，也不會想去用。
- 消息不丟失。用戶發布的一個寶貝消息不能在實時處理的時候給丟了，對吧？更嚴格一點，如果是一個精確數據統計的應用，那么它處理的消息要不多不少才行。這個要求有點高哦。
在2011年Storm開源之前，由於Hadoop的火紅，整個業界都在喋喋不休地談論大數據。Hadoop的高吞吐，海量數據處理的能力使得人們可以方便地處理海量數據。但是，Hadoop的缺點也和它的優點同樣鮮明——延遲大，響應緩慢，運維復雜。
有需求也就有創造，在Hadoop基本奠定了大數據霸主地位的時候，很多的開源項目都是以彌補Hadoop的實時性為目標而被創造出來。而在這個節骨眼上Storm橫空出世了。
Storm帶着流式計算的標簽華麗麗滴出場了，看看它的一些賣點：
- 分布式系統：可橫向拓展,現在的項目不帶個分布式特性都不好意思開源。
- 運維簡單：Storm的部署的確簡單。雖然沒有Mongodb的解壓即用那么簡單，但是它也就是多安裝兩個依賴庫而已。
- 高度容錯：模塊都是無狀態的，隨時宕機重啟。
- 無數據丟失：Storm創新性提出的ack消息追蹤框架和復雜的事務性處理,能夠滿足很多級別的數據處理需求。不過，越高的數據處理需求，性能下降越嚴重。
- 多語言：實際上，Storm的多語言更像是臨時添加上去似的。因為，你的提交部分還是要使用Java實現。
二、簡單認識Storm
Storm是一個免費開源、分布式、高容錯的實時計算系統。Storm令持續不斷的流計算變得容易，彌補了Hadoop批處理所不能滿足的實時要求。Storm經常用於在實時分析、在線機器學習、持續計算、分布式遠程調用和ETL等領域。Storm的部署管理非常簡單，而且，在同類的流式計算工具，Storm的性能也是非常出眾的。Storm對於實時計算的的意義相當於Hadoop對於批處理的意義。Hadoop為我們提供了Map和Reduce原語，使我們對數據進行批處理變的非常的簡單和優美。同樣，Storm也對數據的實時計算提供了簡單Spout和Bolt原語。
Storm適用的場景：
1、流數據處理：Storm可以用來用來處理源源不斷的消息，並將處理之后的結果保存到持久化介質中。
2、分布式RPC：由於Storm的處理組件都是分布式的，而且處理延遲都極低，所以可以Storm可以做為一個通用的分布式RPC框架來使用。
在這個教程里面我們將學習如何創建Topologies, 並且把topologies部署到storm的集群里面去。Java將是我們主要的示范語言， 個別例子會使用python以演示storm的多語言特性。
1、一個Storm集群的基本組件
storm的集群表面上看和hadoop的集群非常像。但是在Hadoop上面你運行的是MapReduce的Job, 而在Storm上面你運行的是Topology。它們是非常不一樣的 — 一個關鍵的區別是： 一個MapReduce Job最終會結束， 而一個Topology永遠運行（除非你顯式的殺掉他）。
在Storm的集群里面有兩種節點： 控制節點(master node)和工作節點(worker node)。
控制節點上面運行一個后台程序：Nimbus， 它的作用類似Hadoop里面的JobTracker。Nimbus負責在集群里面分布代碼，分配工作給機器， 並且監控狀態。
每一個工作節點上面運行一個叫做Supervisor的節點（類似 TaskTracker）。Supervisor會監聽分配給它那台機器的工作，根據需要 啟動/關閉工作進程。每一個工作進程執行一個Topology（類似 Job）的一個子集；一個運行的Topology由運行在很多機器上的很多工作進程 Worker（類似 Child）組成。
storm topology結構
Storm VS MapReduce
Nimbus和Supervisor之間的所有協調工作都是通過一個Zookeeper集群來完成。並且，nimbus進程和supervisor都是快速失敗（fail-fast)和無狀態的。所有的狀態要么在Zookeeper里面， 要么在本地磁盤上。這也就意味着你可以用kill -9來殺死nimbus和supervisor進程， 然后再重啟它們，它們可以繼續工作， 就好像什么都沒有發生過似的。這個設計使得storm不可思議的穩定。
- Nimbus負責在集群里面發送代碼，分配工作給機器，並且監控狀態。全局只有一個。
- Supervisor會監聽分配給它那台機器的工作，根據需要啟動/關閉工作進程Worker。每一個要運行Storm的機器上都要部署一個，並且按照機器的配置設定上面分配的槽位數。
- Zookeeper是Storm重點依賴的外部資源。Nimbus和Supervisor甚至實際運行的Worker都是把心跳保存在Zookeeper上的。Nimbus也是根據Zookeerper上的心跳和任務運行狀況，進行調度和任務分配的。
- Storm提交運行的程序稱為Topology。
- Topology處理的最小的消息單位是一個Tuple，也就是一個任意對象的數組。
- Topology由Spout和Bolt構成。Spout是發出Tuple的結點。Bolt可以隨意訂閱某個Spout或者Bolt發出的Tuple。Spout和Bolt都統稱為component。
2、Topologies
為了在storm上面做實時計算， 你要去建立一些topologies。一個topology就是一個計算節點所組成的圖。Topology里面的每個處理節點都包含處理邏輯， 而節點之間的連接則表示數據流動的方向。
下圖是一個Topology設計的邏輯圖的例子。
3、Stream
Stream是storm里面的關鍵抽象。一個stream是一個沒有邊界的tuple序列。storm提供一些原語來分布式地、可靠地把一個stream傳輸進一個新的stream。比如： 你可以把一個tweets流傳輸到熱門話題的流。
storm提供的最基本的處理stream的原語是spout和bolt。你可以實現Spout和Bolt對應的接口以處理你的應用的邏輯。
spout的流的源頭。比如一個spout可能從Kestrel隊列里面讀取消息並且把這些消息發射成一個流。又比如一個spout可以調用twitter的一個api並且把返回的tweets發射成一個流。
通常Spout會從外部數據源（隊列、數據庫等）讀取數據，然后封裝成Tuple形式，之后發送到Stream中。Spout是一個主動的角色，在接口內部有個nextTuple函數，Storm框架會不停的調用該函數。
bolt可以接收任意多個輸入stream， 作一些處理， 有些bolt可能還會發射一些新的stream。一些復雜的流轉換， 比如從一些tweet里面計算出熱門話題， 需要多個步驟， 從而也就需要多個bolt。 Bolt可以做任何事情: 運行函數， 過濾tuple, 做一些聚合， 做一些合並以及訪問數據庫等等。
Bolt處理輸入的Stream，並產生新的輸出Stream。Bolt可以執行過濾、函數操作、Join、操作數據庫等任何操作。Bolt是一個被動的角色，其接口中有一個execute(Tuple input)方法，在接收到消息之后會調用此函數，用戶可以在此方法中執行自己的處理邏輯。
spout和bolt所組成一個網絡會被打包成topology， topology是storm里面最高一級的抽象（類似 Job）， 你可以把topology提交給storm的集群來運行。topology的結構在Topology那一段已經說過了，這里就不再贅述了。
topology結構
topology里面的每一個節點都是並行運行的。 在你的topology里面， 你可以指定每個節點的並行度， storm則會在集群里面分配那么多線程來同時計算。
一個topology會一直運行直到你顯式停止它。storm自動重新分配一些運行失敗的任務， 並且storm保證你不會有數據丟失， 即使在一些機器意外停機並且消息被丟掉的情況下。
4、數據模型(Data Model)
storm使用tuple來作為它的數據模型。每個tuple是一堆值，每個值有一個名字，並且每個值可以是任何類型， 在我的理解里面一個tuple可以看作一個沒有方法的java對象。總體來看，storm支持所有的基本類型、字符串以及字節數組作為tuple的值類型。你也可以使用你自己定義的類型來作為值類型， 只要你實現對應的序列化器(serializer)。
一個Tuple代表數據流中的一個基本的處理單元，例如一條cookie日志，它可以包含多個Field，每個Field表示一個屬性。
Tuple本來應該是一個Key-Value的Map，由於各個組件間傳遞的tuple的字段名稱已經事先定義好了，所以Tuple只需要按序填入各個Value，所以就是一個Value List。
一個沒有邊界的、源源不斷的、連續的Tuple序列就組成了Stream。
topology里面的每個節點必須定義它要發射的tuple的每個字段。 比如下面這個bolt定義它所發射的tuple包含兩個字段，類型分別是: double和triple。
public class DoubleAndTripleBolt implements IRichBolt { private OutputCollectorBase _collector; @Override public void prepare(Map conf, TopologyContext context, OutputCollectorBase collector) { _collector = collector; } @Override public void execute(Tuple input) { int val = input.getInteger(0); _collector.emit(input,newValues(val*2, val*3)); _collector.ack(input); } @Override public void cleanup() { } @Override public void declare OutputFields(OutputFieldsDeclarer declarer) { declarer.declare(newFields("double","triple")); } }
declareOutputFields方法定義要輸出的字段 ： ["double", "triple"]。這個bolt的其它部分我們接下來會解釋。
5、一個簡單的Topology
讓我們來看一個簡單的topology的例子， 我們看一下storm-starter里面的ExclamationTopology:
TopologyBuilder builder =new TopologyBuilder(); builder.setSpout(1,new TestWordSpout(),10); builder.setBolt(2,new ExclamationBolt(),3) .shuffleGrouping(1); builder.setBolt(3,new ExclamationBolt(),2) .shuffleGrouping(2);
這個Topology包含一個Spout和兩個Bolt。Spout發射單詞， 每個bolt在每個單詞后面加個”!!!”。這三個節點被排成一條線: spout發射單詞給第一個bolt， 第一個bolt然后把處理好的單詞發射給第二個bolt。如果spout發射的單詞是["bob"]和["john"], 那么第二個bolt會發射["bolt!!!!!!"]和["john!!!!!!"]出來。
我們使用setSpout和setBolt來定義Topology里面的節點。這些方法接收我們指定的一個id， 一個包含處理邏輯的對象(spout或者bolt), 以及你所需要的並行度。
這個包含處理的對象如果是spout那么要實現IRichSpout的接口， 如果是bolt，那么就要實現IRichBolt接口.
最后一個指定並行度的參數是可選的。它表示集群里面需要多少個thread來一起執行這個節點。如果你忽略它那么storm會分配一個線程來執行這個節點。
setBolt方法返回一個InputDeclarer對象， 這個對象是用來定義Bolt的輸入。 這里第一個Bolt聲明它要讀取spout所發射的所有的tuple — 使用shuffle grouping。而第二個bolt聲明它讀取第一個bolt所發射的tuple。shuffle grouping表示所有的tuple會被隨機的分發給bolt的所有task。給task分發tuple的策略有很多種，后面會介紹。
如果你想第二個bolt讀取spout和第一個bolt所發射的所有的tuple， 那么你應該這樣定義第二個bolt:
builder.setBolt(3,new ExclamationBolt(),5) .shuffleGrouping(1) .shuffleGrouping(2);
讓我們深入地看一下這個topology里面的spout和bolt是怎么實現的。Spout負責發射新的tuple到這個topology里面來。TestWordSpout從["nathan", "mike", "jackson", "golda", "bertels"]里面隨機選擇一個單詞發射出來。TestWordSpout里面的nextTuple()方法是這樣定義的：
public void nextTuple() { Utils.sleep(100); final String[] words =new String[] {"nathan","mike", "jackson","golda","bertels"}; final Random rand =new Random(); final String word = words[rand.nextInt(words.length)]; _collector.emit(new Values(word)); }
可以看到，實現很簡單。
ExclamationBolt把”!!!”拼接到輸入tuple后面。我們來看下ExclamationBolt的完整實現。
public static class ExclamationBolt implements IRichBolt { OutputCollector _collector; public void prepare(Map conf, TopologyContext context, OutputCollector collector) { _collector = collector; } public void execute(Tuple tuple) { _collector.emit(tuple,new Values(tuple.getString(0) +"!!!")); _collector.ack(tuple); } public void cleanup() { } public void declare OutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } }
prepare方法提供給bolt一個Outputcollector用來發射tuple。Bolt可以在任何時候發射tuple — 在prepare, execute或者cleanup方法里面, 或者甚至在另一個線程里面異步發射。這里prepare方法只是簡單地把OutputCollector作為一個類字段保存下來給后面execute方法使用。
execute方法從bolt的一個輸入接收tuple(一個bolt可能有多個輸入源). ExclamationBolt獲取tuple的第一個字段，加上”!!!”之后再發射出去。如果一個bolt有多個輸入源，你可以通過調用Tuple#getSourceComponent方法來知道它是來自哪個輸入源的。
execute方法里面還有其它一些事情值得一提： 輸入tuple被作為emit方法的第一個參數，並且輸入tuple在最后一行被ack。這些呢都是Storm可靠性API的一部分，后面會解釋。
cleanup方法在bolt被關閉的時候調用， 它應該清理所有被打開的資源。但是集群不保證這個方法一定會被執行。比如執行task的機器down掉了，那么根本就沒有辦法來調用那個方法。cleanup設計的時候是被用來在local mode的時候才被調用(也就是說在一個進程里面模擬整個storm集群), 並且你想在關閉一些topology的時候避免資源泄漏。
最后，declareOutputFields定義一個叫做”word”的字段的tuple。
以local mode運行ExclamationTopology
讓我們看看怎么以local mode運行ExclamationToplogy。
storm的運行有兩種模式: 本地模式和分布式模式. 在本地模式中， storm用一個進程里面的線程來模擬所有的spout和bolt. 本地模式對開發和測試來說比較有用。 你運行storm-starter里面的topology的時候它們就是以本地模式運行的， 你可以看到topology里面的每一個組件在發射什么消息。
在分布式模式下， storm由一堆機器組成。當你提交topology給master的時候， 你同時也把topology的代碼提交了。master負責分發你的代碼並且負責給你的topolgoy分配工作進程。如果一個工作進程掛掉了， master節點會把它重新分配到其它節點。關於如何在一個集群上面運行topology， 你可以看看Running topologies on a production cluster文章。
下面是以本地模式運行ExclamationTopology的代碼:
Config conf =new Config(); conf.setDebug(true); conf.setNumWorkers(2); LocalCluster cluster =new LocalCluster(); cluster.submitTopology("test", conf, builder.createTopology()); Utils.sleep(10000); cluster.killTopology("test"); cluster.shutdown();
首先， 這個代碼定義通過定義一個LocalCluster對象來定義一個進程內的集群。提交topology給這個虛擬的集群和提交topology給分布式集群是一樣的。通過調用submitTopology方法來提交topology， 它接受三個參數：要運行的topology的名字，一個配置對象以及要運行的topology本身。
topology的名字是用來唯一區別一個topology的，這樣你然后可以用這個名字來殺死這個topology的。前面已經說過了， 你必須顯式的殺掉一個topology， 否則它會一直運行。
Conf對象可以配置很多東西， 下面兩個是最常見的：
- TOPOLOGY_WORKERS(setNumWorkers) 定義你希望集群分配多少個工作進程給你來執行這個topology. topology里面的每個組件會被需要線程來執行。每個組件到底用多少個線程是通過setBolt和setSpout來指定的。這些線程都運行在工作進程里面. 每一個工作進程包含一些節點的一些工作線程。比如， 如果你指定300個線程，60個進程， 那么每個工作進程里面要執行6個線程， 而這6個線程可能屬於不同的組件(Spout, Bolt)。你可以通過調整每個組件的並行度以及這些線程所在的進程數量來調整topology的性能。
- TOPOLOGY_DEBUG(setDebug), 當它被設置成true的話， storm會記錄下每個組件所發射的每條消息。這在本地環境調試topology很有用， 但是在線上這么做的話會影響性能的。
運行中的Topology主要由以下三個組件組成的：
Worker processes（工作進程）——Spout/Bolt中運行具體處理邏輯的進程
Executors (threads)（線程、執行器）——物理線程
Tasks（任務）——具體的處理邏輯對象
下圖簡要描述了這3者之間的關系：
storm集群的一個節點可能有一個或者多個工作進程（worker）運行在一個多個拓撲上，一個工作進程執行拓撲的一個子集。工作進程（worker）屬於一個特定的拓撲，並可能為這個拓撲的一個或者多個組件（spout/bolt）運行一個或多個執行器（executor線程）。一個運行中的拓撲包括多個運行在storm集群內多個節點的進程。
1個worker進程執行的是1個topology的子集（注：不會出現1個worker為多個topology服務）。1個worker進程會啟動1個或多個executor線程來執行1個topology的component(spout或bolt)。因此，1個運行中的topology就是由集群中多台物理機上的多個worker進程組成的。
executor是1個被worker進程啟動的單獨線程。每個executor只會運行1個topology的1個component(spout或bolt)的task（注：task可以是1個或多個，storm默認是1個component只生成1個task，executor線程里會在每次循環里順序調用所有task實例）。
task是最終運行spout或bolt中代碼的單元（注：1個task即為spout或bolt的1個實例，executor線程在執行期間會調用該task的nextTuple或execute方法）。topology啟動后，1個component(spout或bolt)的task數目是固定不變的，但該component使用的executor線程數可以動態調整（例如：1個executor線程可以執行該component的1個或多個task實例）。這意味着，對於1個component存在這樣的條件：#threads<=#tasks（即：線程數小於等於task數目）。默認情況下task的數目等於executor線程數目，即1個executor線程只運行1個task。
Spout或者Bolt的Task個數一旦指定之后就不能改變了，而Executor的數量可以根據情況來進行動態的調整。默認情況下# executor = #tasks即一個Executor中運行着一個Task
6、流分組策略(Stream grouping)
流分組策略告訴topology如何在兩個組件之間發送tuple。 要記住， spouts和bolts以很多task的形式在topology里面同步執行。如果從task的粒度來看一個運行的topology， 它應該是這樣的:
從task角度來看topology
當Bolt A的一個task要發送一個tuple給Bolt B， 它應該發送給Bolt B的哪個task呢？
stream grouping專門回答這種問題的。在我們深入研究不同的stream grouping之前， 讓我們看一下storm-starter里面的另外一個topology。WordCountTopology讀取一些句子， 輸出句子里面每個單詞出現的次數.
TopologyBuilder builder =new TopologyBuilder(); builder.setSpout(1,new RandomSentenceSpout(),5); builder.setBolt(2,new SplitSentence(),8) .shuffleGrouping(1); builder.setBolt(3,new WordCount(),12) .fieldsGrouping(2,new Fields("word"));
SplitSentence對於句子里面的每個單詞發射一個新的tuple, WordCount在內存里面維護一個單詞->次數的mapping， WordCount每收到一個單詞， 它就更新內存里面的統計狀態。
有好幾種不同的stream grouping:
- 最簡單的grouping是shuffle grouping, 它隨機發給任何一個task。上面例子里面RandomSentenceSpout和SplitSentence之間用的就是shuffle grouping, shuffle grouping對各個task的tuple分配的比較均勻。
- 一種更有趣的grouping是fields grouping, SplitSentence和WordCount之間使用的就是fields grouping, 這種grouping機制保證相同field值的tuple會去同一個task， 這對於WordCount來說非常關鍵，如果同一個單詞不去同一個task， 那么統計出來的單詞次數就不對了。
fields grouping是stream合並，stream聚合以及很多其它場景的基礎。在背后呢， fields grouping使用的一致性哈希來分配tuple的。
還有一些其它類型的stream grouping. 你可以在Concepts一章里更詳細的了解。
下面是一些常用的 “路由選擇” 機制：
Storm的Grouping即消息的Partition機制。當一個Tuple被發送時，如何確定將它發送個某個（些）Task來處理？？
l ShuffleGrouping：隨機選擇一個Task來發送。
l FiledGrouping：根據Tuple中Fields來做一致性hash，相同hash值的Tuple被發送到相同的Task。
l AllGrouping：廣播發送，將每一個Tuple發送到所有的Task。
l GlobalGrouping：所有的Tuple會被發送到某個Bolt中的id最小的那個Task。
l NoneGrouping：不關心Tuple發送給哪個Task來處理，等價於ShuffleGrouping。
l DirectGrouping：直接將Tuple發送到指定的Task來處理。, 這是一種比較特別的分組方法，用這種分組意味着消息的發送者指定由消息接收者的哪個task處理這個消息。 只有被聲明為Direct Stream的消息流可以聲明這種分組方法。而且這種消息tuple必須使用emitDirect方法來發射。消息處理者可以通過 TopologyContext來獲取處理它的消息的taskid (OutputCollector.emit方法也會返回taskid)。
7、使用別的語言來定義Bolt
Bolt可以使用任何語言來定義。用其它語言定義的bolt會被當作子進程(subprocess)來執行， storm使用JSON消息通過stdin/stdout來和這些subprocess通信。這個通信協議是一個只有100行的庫， storm團隊給這些庫開發了對應的Ruby, Python和Fancy版本。
下面是WordCountTopology里面的SplitSentence的定義:
public static class SplitSentence extends ShellBolt implements IRichBolt { public SplitSentence() { super("python","splitsentence.py"); } public void declareOutputFields(OutputFieldsDeclarer declarer) { declarer.declare(new Fields("word")); } }
SplitSentence繼承自ShellBolt並且聲明這個Bolt用python來運行，並且參數是: splitsentence.py。下面是splitsentence.py的定義:
import storm class SplitSentenceBolt(storm.BasicBolt): def process(self, tup): words=tup.values[0].split(" ") for word in words: storm.emit([word]) SplitSentenceBolt().run()
8、可靠的消息處理
在這個教程的前面，我們跳過了有關tuple的一些特征。這些特征就是storm的可靠性API： storm如何保證spout發出的每一個tuple都被完整處理。看看《storm如何保證消息不丟失》以更深入了解storm的可靠性API.
Storm允許用戶在Spout中發射一個新的源Tuple時為其指定一個MessageId，這個MessageId可以是任意的Object對象。多個源Tuple可以共用同一個MessageId，表示這多個源Tuple對用戶來說是同一個消息單元。Storm的可靠性是指Storm會告知用戶每一個消息單元是否在一個指定的時間內被完全處理。完全處理的意思是該MessageId綁定的源Tuple以及由該源Tuple衍生的所有Tuple都經過了Topology中每一個應該到達的Bolt的處理。
在Spout中由message 1綁定的tuple1和tuple2分別經過bolt1和bolt2的處理，然后生成了兩個新的Tuple，並最終流向了bolt3。當bolt3處理完之后，稱message 1被完全處理了。
Storm中的每一個Topology中都包含有一個Acker組件。Acker組件的任務就是跟蹤從Spout中流出的每一個messageId所綁定的Tuple樹中的所有Tuple的處理情況。如果在用戶設置的最大超時時間內這些Tuple沒有被完全處理，那么Acker會告訴Spout該消息處理失敗，相反則會告知Spout該消息處理成功。
那么Acker是如何記錄Tuple的處理結果呢？？
A xor A = 0.
A xor B…xor B xor A = 0，其中每一個操作數出現且僅出現兩次。
在Spout中，Storm系統會為用戶指定的MessageId生成一個對應的64位的整數，作為整個Tuple Tree的RootId。RootId會被傳遞給Acker以及后續的Bolt來作為該消息單元的唯一標識。同時，無論Spout還是Bolt每次新生成一個Tuple時，都會賦予該Tuple一個唯一的64位整數的Id。
當Spout發射完某個MessageId對應的源Tuple之后，它會告訴Acker自己發射的RootId以及生成的那些源Tuple的Id。而當Bolt處理完一個輸入Tuple並產生出新的Tuple時，也會告知Acker自己處理的輸入Tuple的Id以及新生成的那些Tuple的Id。Acker只需要對這些Id進行異或運算，就能判斷出該RootId對應的消息單元是否成功處理完成了。
摘抄自：