淺談分佈式計算的開發與實現(二)

實時計算

淺談分佈式計算的開發與實現(一)

接上篇，離線計算是對已經入庫的數據進行計算，在查詢時對批量數據進行檢索、磁盤讀取展現。 而實時計算是在數據產生時就對其進行計算，而後實時展現結果，通常是秒級。 舉個例子來講，若是有個大型網站，要實時統計用戶的搜索內容，這樣就能計算出熱點新聞及突發事件了。 按照之前離線計算的作法是不能知足的，須要使用到實時計算。

小明做爲有理想、有追求的程序員開始設計其解決方案了，主要分三部分。

• 每當搜索內容的數據產生時，先把數據收集到消息隊列，因爲其數據量較大，以使用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。 示例以下：

<pre style="margin:0px; padding:0px; white-space:pre-wrap; overflow-wrap:break-word; font-family:&quot; Courier New&quot; !important; font-size:12px !important; ">
public class CalcPriceSpout:BaseRichSpout {
	private SpoutCollector Collector;
	public override void NexData() {
	//讀取各類數據源，Redis、消息隊列、數據庫等
	Collector.emit("消息")
}
}</pre>
複製代碼

代碼中NexData是storm的核心方法，它一直被storm循環調用着， 在方法裏咱們實時讀取kafka的消息，而後把消息經過Collector組件發射到各個計算節點裏，它相似小和尚中的Master。 這樣應用每產生一條數據，會實時收集到kafka，而後被NextData消費，發射到節點開始計算。 NextData讀取的消息時在內存中，而後直接經過網絡流動到節點機器上的內存中開始計算，不會持久化到磁盤上。

由於速度比較快，因此叫實時計算，也有叫持續計算，意思是能夠很是快的一直進行計算，至於叫什麼均可以。

流式計算

主流的流式計算有S四、StreamBase、Borealis，其storm也具備流式計算的特性。 流式計算是指「數據能像液體水同樣不斷的在各個節點間流動，每一個節點均可以對「數據(液體水)」進行計算，而後產生新的數據，繼續像水同樣流動」。如圖：

圖中Spout就是水龍頭，它不斷的經過NextData產生數據，而後流動各個Bolt中。 Bolt是各個計算節點上的計算邏輯，它拿到數據後開始計算，完成後流向另一個，直到完成。 其Bolt也能夠是任意個，這比Mapreduce只能分紅Map、Reduce兩部分好多了。 這樣能夠在BlotA中計算中間值，而後經過這個中間值去任意數據源拉取數據後，在流動到下一步處理邏輯中， 這個中間值直接在內存中，經過網絡流動BlotB上。 其大大增長了其適用範圍和靈活度，Spout和bolt的數據流動構成了一個有向無環圖。 Bolt示例代碼以下。

<pre style="margin:0px; padding:0px; white-space:pre-wrap; overflow-wrap:break-word; font-family:&quot; Courier New&quot; !important; font-size:12px !important; ">
public class CalcProductPriceBolt:BaseRichBolt {
	private BoltCollector Collector;
	public override void Execute(Tuple<string,string> input) {
	//Result=計算計算計算。 //Collector.Emit("Reulst");
	流動到另一個節點
}
}</pre>
複製代碼

數據流動圖：

概括總結

結合上篇，發現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組件外都是多節點點部署，掛了某一節點，不會對任務計算有所影響。