Storm 簡介

場景

伴隨著信息科技日新月異的發展，信息呈現出爆發式的膨脹，人們獲取信息的途徑也更加多樣、更加便捷，同時對于信息的時效性要求也越來越高。舉個搜索場景中的例子，當一個賣家發布了一條寶貝信息時，他希望的當然是這個寶貝馬上就可以被賣家搜索出來、點擊、購買啦，相反，如果這個寶貝要等到第二天或者更久才可以被搜出來，估計這個大哥就要罵娘了。再舉一個推薦的例子，如果用戶昨天在淘寶上買了一雙襪子，今天想買一副泳鏡去游泳，但是卻發現系統在不遺余力地給他推薦襪子、鞋子，根本對他今天尋找泳鏡的行為視而不見，估計這哥們心里就會想推薦你妹呀。其實稍微了解點背景知識的碼農們都知道，這是因為后臺系統做的是每天一次的全量處理，而且大多是在夜深人靜之時做的，那么你今天白天做的事情當然要明天才能反映出來啦。

實現一個實時計算系統

全量數據處理使用的大多是鼎鼎大名的hadoop或者hive，作為一個批處理系統，hadoop以其吞吐量大、自動容錯等優點，在海量數據處理上得到了廣泛的使用。但是，hadoop不擅長實時計算，因為它天然就是為批處理而生的，這也是業界一致的共識。否則最近這兩年也不會有s4, Storm ,puma這些實時計算系統如雨后春筍般冒出來啦。先拋開s4, Storm ,puma這些系統不談，我們首先來看一下，如果讓我們自己設計一個實時計算系統，我們要解決哪些問題。

低延遲。都說了是實時計算系統了，延遲是一定要低的。

高性能。性能不高就是浪費機器，浪費機器是要受批評的哦。

分布式。系統都是為應用場景而生的，如果你的應用場景、你的數據和計算單機就能搞定，那么不用考慮這些復雜的問題了。我們所說的是單機搞不定的情況。

可擴展。伴隨著業務的發展，我們的數據量、計算量可能會越來越大，所以希望這個系統是可擴展的。

容錯。這是分布式系統中通用問題。一個節點掛了不能影響我的應用。

好，如果僅僅需要解決這5個問題，可能會有無數種方案，而且各有千秋，隨便舉一種方案，使用消息隊列+分布在各個機器上的工作進程就ok啦。我們再繼續往下看。

容易在上面開發應用程序。親，你設計的系統需要應用程序開發人員考慮各個處理組件的分布、消息的傳遞嗎？如果是，那有點麻煩啊，開發人員可能會用不好，也不會想去用。

消息不丟失。用戶發布的一個寶貝消息不能在實時處理的時候給丟了，對吧？更嚴格一點，如果是一個精確數據統計的應用，那么它處理的消息要不多不少才行。這個要求有點高哦。

消息嚴格有序。有些消息之間是有強相關性的，比如同一個寶貝的更新和刪除操作消息，如果處理時搞亂順序完全是不一樣的效果了。

不知道大家對這些問題是否都有了自己的答案，下面讓我們帶著這些問題，一起來看一看 Storm 的解決方案吧。

Storm 是什么

流數據處理。 Storm 可以用來處理源源不斷流進來的消息，處理之后將結果寫入到某個存儲中去。

分布式rpc。由于 Storm 的處理組件是分布式的，而且處理延遲極低，所以可以作為一個通用的分布式rpc框架來使用。當然，其實我們的搜索引擎本身也是一個分布式rpc系統。

說了半天，好像都是很玄乎的東西，下面我們開始具體講解 Storm 的基本概念和它內部的一些實現原理吧。

Storm 的基本概念

首先我們通過一個 Storm 和 hadoop 的對比來了解 Storm 中的基本概念。

接下來我們再來具體看一下這些概念。

Nimbus：負責資源分配和任務調度。

Supervisor：負責接受nimbus分配的任務，啟動和停止屬于自己管理的worker進程。

Worker：運行具體處理組件邏輯的進程。

Task：worker中每一個 Spout /bolt的線程稱為一個task. 在 Storm 0.8之后，task不再與物理線程對應，同一個 Spout /bolt的task可能會共享一個物理線程，該線程稱為executor。

下面這個圖描述了以上幾個角色之間的關系。

Topology： Storm 中運行的一個實時應用程序，因為各個組件間的消息流動形成邏輯上的一個拓撲結構。

Spout：在一個topology中產生源數據流的組件。通常情況下 Spout 會從外部數據源中讀取數據，然后轉換為topology內部的源數據。 Spout 是一個主動的角色，其接口中有個nextTuple()函數， Storm 框架會不停地調用此函數，用戶只要在其中生成源數據即可。

Bolt：在一個topology中接受數據然后執行處理的組件。Bolt可以執行過濾、函數操作、合并、寫數據庫等任何操作。Bolt是一個被動的角色，其接口中有個execute(Tuple input)函數,在接受到消息后會調用此函數，用戶可以在其中執行自己想要的操作。

Tuple：一次消息傳遞的基本單元。本來應該是一個key-value的map，但是由于各個組件間傳遞的tuple的字段名稱已經事先定義好，所以tuple中只要按序填入各個value就行了，所以就是一個value list.

Stream：源源不斷傳遞的tuple就組成了stream。

Stream Grouping：即消息的partition方法。 Storm 中提供若干種實用的grouping方式，包括shuffle, fields hash, all, global, none, direct和localOrShuffle等。

相比于s4, puma等其他實時計算系統， Storm 最大的亮點在于其記錄級容錯和能夠保證消息精確處理的事務功能。下面就重點來看一下這兩個亮點的實現原理。

Storm 記錄級容錯的基本原理

首先來看一下什么叫做記錄級容錯？ Storm 允許用戶在 Spout 中發射一個新的源tuple時為其指定一個message id, 這個message id可以是任意的object對象。多個源tuple可以共用一個message id，表示這多個源 tuple對用戶來說是同一個消息單元。 Storm 中記錄級容錯的意思是說， Storm 會告知用戶每一個消息單元是否在指定時間內被完全處理了。那什么叫做完全處理呢，就是該message id綁定的源tuple及由該源tuple后續生成的tuple經過了topology中每一個應該到達的bolt的處理。舉個例子，在 Spout 由message 1綁定的tuple1和tuple2經過了bolt1和bolt2的處理生成兩個新的tuple，并最終都流向了bolt3。當這個過程完成處理完時，稱message 1被完全處理了。

在 Storm 的topology中有一個系統級組件，叫做acker。這個acker的任務就是追蹤從 Spout 中流出來的每一個message id綁定的若干tuple的處理路徑，如果在用戶設置的最大超時時間內這些tuple沒有被完全處理，那么acker就會告知 Spout 該消息處理失敗了，相反則會告知 Spout 該消息處理成功了。在剛才的描述中，我們提到了“記錄tuple的處理路徑”，如果曾經嘗試過這么做的同學可以仔細地思考一下這件事的復雜程度。但是 Storm 中卻是使用了一種非常巧妙的方法做到了。在說明這個方法之前，我們來復習一個數學定理：

A xor A = 0.

A xor B xor … xor B xor A = 0，其中每一個操作數出現且僅出現兩次。

Storm 中使用的巧妙方法就是基于這個定理。具體過程是這樣的：在 Spout 中系統會為用戶指定的 message id 生成一個對應的64位整數，作為一個 root id。root id 會傳遞給acker及后續的bolt作為該消息單元的唯一標識。同時無論是 Spout 還是bolt每次新生成一個tuple的時候，都會賦予該tuple一個64位的整數的id。 Spout 發射完某個message id對應的源tuple之后，會告知acker自己發射的root id及生成的那些源tuple的id。而bolt呢，每次接受到一個輸入tuple處理完之后，也會告知acker自己處理的輸入tuple的id及新生成的那些tuple的id。Acker只需要對這些id做一個簡單的異或運算，就能判斷出該root id對應的消息單元是否處理完成了。下面通過一個圖示來說明這個過程。

spout中綁定message 1生成了兩個源tuple，id分別是0010和1011。

bolt1處理tuple 0010時生成了一個新的tuple，id為0110。

bolt3中接收到tuple 0110和tuple 0111，沒有生成新的tuple。

可能有些細心的同學會發現，容錯過程存在一個可能出錯的地方，那就是，如果生成的tuple id并不是完全各異的，acker可能會在消息單元完全處理完成之前就錯誤的計算為0。這個錯誤在理論上的確是存在的，但是在實際中其概率是極低極低的，完全可以忽略。

Storm 的事務拓撲

事務拓撲(transactional topology)是storm0.7引入的特性，在最近發布的0.8版本中已經被封裝為Trident，提供了更加便利和直觀的接口。因為篇幅所限，在此對事務拓撲做一個簡單的介紹。

事務拓撲的目的是為了滿足對消息處理有著極其嚴格要求的場景，例如實時計算某個用戶的成交筆數，要求結果完全精確，不能多也不能少。Storm的事務拓撲是完全基于它底層的spout/bolt/acker原語實現的，通過一層巧妙的封裝得出一個優雅的實現。個人覺得這也是storm最大的魅力之一。

事務拓撲簡單來說就是將消息分為一個個的批(batch)，同一批內的消息以及批與批之間的消息可以并行處理，另一方面，用戶可以設置某些bolt為committer，storm可以保證committer的finishBatch()操作是按嚴格不降序的順序執行的。用戶可以利用這個特性通過簡單的編程技巧實現消息處理的精確。

Storm 在淘寶

由于storm的內核是clojure編寫的(不過大部分的拓展工作都是java編寫的)，為我們理解它的實現帶來了一定的困難，好在大部分情況下storm都比較穩定，當然我們也在盡力熟悉clojure的世界。我們在使用storm時通常都是選擇java語言開發應用程序。

在淘寶，storm被廣泛用來進行實時日志處理，出現在實時統計、實時風控、實時推薦等場景中。一般來說，我們從類kafka的metaQ或者基于hbase的timetunnel中讀取實時日志消息，經過一系列處理，最終將處理結果寫入到一個分布式存儲中，提供給應用程序訪問。我們每天的實時消息量從幾百萬到幾十億不等，數據總量達到TB級。對于我們來說，storm往往會配合分布式存儲服務一起使用。在我們正在進行的個性化搜索實時分析項目中，就使用了timetunnel + hbase + storm + ups的架構，每天處理幾十億的用戶日志信息，從用戶行為發生到完成分析延遲在秒級。

Storm 的未來

Storm0.7系列的版本已經在各大公司得到了廣泛使用，最近發布的0.8版本中引入了State，使得其從一個純計算框架演變成了一個包含存儲和計算的實時計算新利器，還有剛才提到的Trident，提供更加友好的接口，同時可定制scheduler的特性也為其針對不同的應用場景做優化提供了更便利的手段，也有人已經在基于storm的實時ql(query language)上邁出了腳本。在服務化方面，storm一直在朝著融入mesos框架的方向努力。同時，storm也在實現細節上不斷地優化，使用很多優秀的開源產品，包括kryo, Disruptor, curator等等?？梢韵胂螅攕torm發展到1.0版本時，一定是一款無比杰出的產品，讓我們拭目以待，當然，最好還是參與到其中去吧，同學們。

Storm簡介

任務調度 分布式

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。