閱讀目錄:
- 實時計算
- storm簡介
- 流式計算
- 歸納總結
- 高容錯性
接上篇，離線計算是對已經入庫的資料進行計算，在查詢時對批量資料進行檢索、磁盤讀取展示。 而實時計算是在資料産生時就對其進行計算，然後實時展示結果，一般是秒級。 舉個例子來說，如果有個大型網站，要實時統計使用者的搜尋内容，這樣就能計算出熱點新聞及突發事件了。 按照以前離線計算的做法是不能滿足的，需要使用到實時計算。
小明作為有理想、有追求的程式員開始設計其解決方案了，主要分三部分。
- 每當搜尋内容的資料産生時，先把資料收集到消息隊列，由于其資料量較大，以使用kafka為例。 這個收集過程是一直持續的，資料不斷産生然後不斷流入到kafka中。
- 要有一個能持續計算的架構，一旦收集到資料，計算系統能實時收到資料，根據業務邏輯開始計算，然後不斷産生需要的結果，這裡以storm為例。
- 根據結果進行實時展示并入庫， 可以一邊展示一邊入庫，對外提供實時查詢的服務。這裡的入庫可以是基于記憶體的Redis、MongoDB，也可是基于磁盤的HBase、Mysql、SqlServer等。
其流程圖如下:
通常都介紹Storm是一個分布式的、高容錯的實時計算系統。 “分布式”是把資料分布到多台上進行計算，“高容錯”下面談，這裡主要細節介紹下“實時計算”的實作。
storm有個角色叫topology，它類似mapreduce的job，是一個完整的業務計算任務抽象。 上章談到hadoop的缺點在于資料源單一依賴HDFS，storm中Spout角色的出現解決了這個問題。 在Spout内部我們可以讀取任意資料源的資料，比如Redis、消息隊列、資料庫等等。 而且spout可以是多個，這樣更好的分類，比如可以SpoutA讀取kafka，SpoutB讀取Redis。 示例如下：
public class CalcPriceSpout : BaseRichSpout
{
private SpoutCollector Collector;
public override void NexData()
{
//讀取各種資料源，Redis、消息隊列、資料庫等
Collector.emit("消息")
}
} 代碼中NexData是storm的核心方法，它一直被storm循環調用着， 在方法裡我們實時讀取kafka的消息，然後把消息通過Collector元件發射到各個計算節點裡，它類似小和尚中的Master。 這樣應用每産生一條資料，會實時收集到kafka，然後被NextData消費，發射到節點開始計算。 NextData讀取的消息時在記憶體中，然後直接通過網絡流動到節點機器上的記憶體中開始計算，不會持久化到磁盤上。
因為速度比較快，是以叫實時計算，也有叫持續計算，意思是可以非常快的一直進行計算，至于叫什麼都可以。
主流的流式計算有S4、StreamBase、Borealis，其storm也具有流式計算的特性。 流式計算是指“資料能像液體水一樣不斷的在各個節點間流動，每個節點都可以對“資料(液體水)”進行計算，然後産生新的資料，繼續像水一樣流動”。如圖：
圖中Spout就是水龍頭，它不斷的通過NextData産生資料，然後流動各個Bolt中。 Bolt是各個計算節點上的計算邏輯，它拿到資料後開始計算，完成後流向另外一個，直到完成。 其Bolt也可以是任意個，這比Mapreduce隻能分成Map、Reduce兩部分好多了。 這樣可以在BlotA中計算中間值，然後通過這個中間值去任意資料源拉取資料後，在流動到下一步處理邏輯中， 這個中間值直接在記憶體中，通過網絡流動BlotB上。 其大大增加了其适用範圍和靈活度，Spout和bolt的資料流動構成了一個有向無環圖。 Bolt示例代碼如下。
public class CalcProductPriceBolt : BaseRichBolt
{
private BoltCollector Collector;
public override void Execute(Tuple<string, string> input)
{
//Result=計算計算計算。
//Collector.Emit("Reulst"); 流動到另外一個節點
}
} 資料流動圖：
結合上篇，發現Hadoop離線計算的計算要求是把業務邏輯包上傳到平台上，資料導入到HDFS上，這樣才能進行計算。 其産生的結果資料是展示之前就計算好的，另外它的計算是按批次來的，比如很多公司的報表，都是每天淩晨開始計算前一天的資料，以便于展示。 其資料是不動的，計算邏輯也是不動的。
Storm的流式計算同樣是把計算邏輯包上傳到平台上，由平台排程，計算邏輯是不動的。 但資料可以是任意來源的，不斷在計算節點進行流動。 也即是說在資料産生的時刻，就開始進行流動計算，它展示的結果資料是實時變化的。 其資料是流動的，計算邏輯是不動的。storm把産生的每條資料當成一個消息來處理，其内部也是通過消息隊列元件zeromq來完成的。
storm提供了各級别的可靠性保證，一消息從Spout流動到boltA，在流動boltB， 那storm會通過唯一值不斷異或的設計去監測這個消息的完成情況，這個監測是一個和業務邏輯類似的bolt，不過它是有storm自身實作的，叫Acker，它的任務就是接收各個消息任務的完成狀态，然後告訴Spout這個消息是否已經完全處理。下面是幾種異常處理情況：
- BoltB所在的節點挂了或消息異常，那麼這條消息就沒有處理完，Spout可在逾時後重新發射該資料即可。
- Acker所在節點挂了後，即目前節點監控的消息完全情況，會全部丢失，Spout會在消息逾時做後續處理。
- 如果Spout所在節點挂了，那Spout發射的資料也會全部丢失， 這時可在消息隊列中設定逾時時間，如果沒有一直沒對消息進行Ack的話，那麼這條消息會重新讓其他的Spout重新接收到。這部分需要單獨在消息隊列中配置，另外storm消息的Ack确認對性能有一定影響，可根據消息的重要性是否要開啟它。
- 如果storm平台級别的元件挂了，平台會嘗試重新開機失敗的元件，storm除nimbus元件外都是多節點點部署，挂了某一節點，不會對任務計算有所影響。
下篇寫消息保證機制及改造小和尚的設計。
作者：蘑菇先生
出處： http://mushroom.cnblogs.com/
本文版權歸作者和部落格園共有，歡迎轉載。未經作者同意下，必須在文章頁面明顯标出原文連結及作者，否則保留追究法律責任的權利。
如果您認為這篇文章還不錯或者有所收獲，可以點選右下角的
【推薦】按鈕，因為你的支援是我繼續寫作，分享的最大動力！