閱讀目錄:
實時計算
接上篇，離線計算是對已經入庫的數據進行計算，在查詢時對批量數據進行檢索、磁盤讀取展示。 而實時計算是在數據產生時就對其進行計算，然后實時展示結果，一般是秒級。 舉個例子來說，如果有個大型網站，要實時統計用戶的搜索內容，這樣就能計算出熱點新聞及突發事件了。 按照以前離線計算的做法是不能滿足的，需要使用到實時計算。
小明作為有理想、有追求的程序員開始設計其解決方案了，主要分三部分。
- 每當搜索內容的數據產生時，先把數據收集到消息隊列，由於其數據量較大，以使用kafka為例。 這個收集過程是一直持續的，數據不斷產生然后不斷流入到kafka中。
- 要有一個能持續計算的框架，一旦收集到數據，計算系統能實時收到數據，根據業務邏輯開始計算，然后不斷產生需要的結果，這里以storm為例。
- 根據結果進行實時展示並入庫， 可以一邊展示一邊入庫，對外提供實時查詢的服務。這里的入庫可以是基於內存的Redis、MongoDB，也可是基於磁盤的HBase、Mysql、SqlServer等。
其流程圖如下:
storm簡介
通常都介紹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組件外都是多節點點部署，掛了某一節點，不會對任務計算有所影響。
下篇寫消息保證機制及改造小和尚的設計。