最近利用閑暇時間，又重新研讀了一下Storm。認真對比了一下Hadoop，前者更擅長的是，實時流式資料處理，後者更擅長的是基于HDFS，通過MapReduce方式的離線資料分析計算。對于Hadoop，本身不擅長實時的資料分析處理。兩者的共同點都是分布式的架構，而且，都類似有主/從關系的概念。本文中我就不具體闡述Storm叢集和Zookeeper叢集如何部署的問題，我想通過一個實際的案例切入，分析一下如何利用Storm，完成實時分析處理資料的。
Storm本身是Apache托管的開源的分布式實時計算系統，它的前身是Twitter Storm。在Storm問世以前，處理海量的實時資料資訊，大部分是類似于使用消息隊列，加上工作程序/線程的方式。這使得建構這類的應用程式，變得異常的複雜。很多的業務邏輯中，你不得不考慮消息的發送和接收，線程之間的并發控制等等問題。而其中的業務邏輯可能隻是占據整個應用的一小部分，而且很難做到業務邏輯的解耦。但是Storm的出現改變了這種局面，它首先抽象出資料流Stream的抽象概念，一個Stream指的是tuples組成的無邊界的序列。後面又繼續提出Spouts、Bolts的概念。Spouts在Storm裡面是資料源，專門負責生成流。而Bolts則是以流作為輸入，并重新生成流作為輸出，并且Bolts還會繼續指定它輸入的流應該如何劃分。最後Storm是通過拓撲（Topology）這種抽象概念，組織起若幹個Spouts、Bolts構成的分布式資料處理網絡。Storm設計的時候，就有意的把Spouts、Bolts組成的拓撲（Topology）網絡通過Thrift服務方式進行封裝，這個做法，使得Storm的Spouts、Bolts元件可以通過目前主流的任意語言實作，使得整個架構的相容性和擴充性更加的優秀。
在Storm裡面拓撲（Topology）的概念，非常類似Hadoop裡面MapReduce的Job的概念。不同的是Storm的拓撲（Topology）隻要你啟動了，它就會一直運作下去，除非你kill掉；而MapReduce的Job最終它是會結束的。基于這樣的模式，使得Storm非常适合處理實時性的資料分析，持續計算，DRPC（分布式RPC）等。
好了，我就結合實際的案例，設計分析一下，如何利用Storm改善應用的處理性能。
移動公司的垃圾短信監控平台，實時地上傳每個省的疑似垃圾短信使用者的垃圾短信内容檔案，每個省則根據檔案中垃圾短信的内容，解析過濾出，包含指定敏感關鍵字的垃圾短信進行入庫。被入庫的垃圾短信使用者被列為敏感使用者，是重點監控對象，畢竟亂發這些垃圾短信是非常不對的。垃圾短信監控平台生成的檔案速度非常驚人，原來的傳統做法是，根據每個省的每一個地市，對應一個獨立應用，串行化地解析、過濾敏感關鍵字，來進行入庫處理。但是，從現狀來看，程式處理的性能并不高效，常常造成檔案積壓，沒有及時處理入庫。
現在，我們就通過Storm，來重新梳理、組織一下上述的應用場景。
首先，我先說明一下，該案例中，Storm叢集和Zookeeper叢集的部署情況，如下圖所示：
Nimbus對應的主機是192.168.95.134是Storm主節點，其餘兩台從節點Supervisor對應的主機分别是192.168.95.135（主機名：slave1）、192.168.95.136（主機名：slave2）。同樣的，Zookeeper叢集也是部署在上述節點上。Storm叢集和Zookeeper叢集會互相通信，因為Storm就是基于Zookeeper的。然後先啟動每個節點的Zookeeper服務，其次分别啟動Storm的Nimbus、Supervisor服務。具體可以到Storm安裝的bin目錄下面啟動服務，啟動指令分别為storm nimbus > /dev/null 2 > &1 &和storm supervisor > /dev/null 2 > &1 &。然後用jps觀察啟動的效果。沒有問題的話，在Nimbus服務對應的主機上啟動Storm UI監控對應的服務，在Storm安裝目錄的bin目錄輸入指令：storm ui >/dev/null 2>&1 &。然後打開浏覽器輸入：http://{Nimbus服務對應的主機ip}:8080，這裡就是輸入：http://192.168.95.134:8080/。觀察Storm叢集的部署情況，如下圖所示：
可以發現，我們的Storm的版本是0.9.5，它的從節點（Supervisor）有2個，分别是slave1、slave2。一共的woker的數量是8個（Total slots）。Storm叢集我們已經部署完畢，也啟動成功了。現在我們就利用Storm的方式，來重新改寫一下這種敏感資訊實時監控過濾的應用。首先看下，Storm方式的拓撲結構圖：
其中的SensitiveFileReader-591、SensitiveFileReader-592（使用者短信采集器，分地市）代表的是Storm中的Spouts元件，表示一個資料的源頭，這裡是表示從伺服器的指定目錄下，讀取疑似垃圾短信使用者的垃圾短信内容檔案。當然Spouts的元件你可以根據實際的需求，擴充出許多Spouts。
然後讀取出檔案中每一行的内容之後，就是分析檔案的内容元件了，這裡是指：SensitiveFileAnalyzer（監控短信内容拆解分析），它負責分析出檔案的格式内容。
為了簡單示範起見，我這裡定義檔案的格式為如下内容（随便寫一個例子）：home_city=591&user_id=5911000&msisdn=10000&sms_content=abc-slave1。每個列之間用&進行連接配接。其中home_city=591表示疑似垃圾短信的使用者歸屬地市編碼，591表示福州、592表示廈門；user_id=5911000表示疑似垃圾短信的使用者辨別；msisdn=10000表示疑似垃圾短信的使用者手機号碼；sms_content=abc-slave1代表的就是垃圾短信的内容了。SensitiveFileAnalyzer代表的就是Storm中的Bolt元件，用來處理Spouts“流”出的資料。
最後，就是我們根據解析好的資料，比對業務規定的敏感關鍵字，進行過濾入庫了。這裡我們是把過濾好的資料存入MySQL資料庫中。負責這項任務的元件是：SensitiveBatchBolt（敏感資訊采集處理），當然它也是Storm中的Bolt元件。好了，以上就是完整的Storm拓撲（Topology）結構了。
現在，我們對于整個敏感資訊采集過濾監控的拓撲結構，有了一個整體的了解之後，我們再來看下如何具體編碼實作！先來看下整個工程的代碼層次結構，它如下圖所示：
首先來看下，我們定義的敏感使用者的資料結構RubbishUsers，假設，我們要過濾的敏感使用者的短信内容中，要包含“racketeer”、“Bad”等敏感關鍵字。具體代碼如下：
現在，我們看下敏感資訊資料源元件SensitiveFileReader的具體實作，它負責從伺服器的指定目錄下面，讀取疑似垃圾短信使用者的垃圾短信内容檔案，然後把每一行的資料，發送給下一個處理的Bolt（SensitiveFileAnalyzer），每個檔案全部發送結束之後，在目前目錄中，把原檔案重命名成字尾bak的檔案（當然，你可以重建立立一個備份目錄，專門用來存儲這種處理結束的檔案），SensitiveFileReader的具體實作如下：
監控短信内容拆解分析器SensitiveFileAnalyzer，這個Bolt元件，接收到資料源SensitiveFileReader的資料之後，就按照上面定義的格式，對檔案中每一行的内容進行解析，然後把解析完畢的内容，繼續發送給下一個Bolt元件：SensitiveBatchBolt（敏感資訊采集處理）。現在，我們來看下SensitiveFileAnalyzer這個Bolt元件的實作：
最後一個Bolt元件SensitiveBatchBolt（敏感資訊采集處理）根據上遊Bolt元件SensitiveFileAnalyzer發送過來的資料，然後跟業務規定的敏感關鍵字進行比對，如果比對成功，說明這個使用者，就是我們要重點監控的使用者，我們把他，通過hibernate采集到MySQL資料庫，統一管理。最後要說明的是，SensitiveBatchBolt元件還實作了一個監控的功能，就是定期列印出，我們已經采集到的敏感資訊使用者資料。現在給出SensitiveBatchBolt的實作：
由于是通過hibernate入庫到MySQL，是以給出hibernate配置，首先是：hibernate.cfg.xml
對應的ORM映射配置檔案rubbish-users.hbm.xml内容如下：
最後，還是通過Spring把hibernate內建起來，資料庫連接配接池用的是：DBCP。對應的Spring配置檔案jdbc-hibernate-bean.xml的内容如下：
到此為止，我們已經完成了敏感資訊實時監控的所有的Storm元件的開發。現在，我們來完成Storm的拓撲（Topology），由于拓撲（Topology）又分為本地拓撲和分布式拓撲，是以封裝了一個工具類StormRunner（拓撲執行器），對應的代碼如下：
好了，現在我們把上面所有的Spouts/Bolts拼接成“拓撲”（Topology）結構，我們這裡用的是分布式拓撲，來進行部署運作。具體的SensitiveTopology（敏感使用者監控Storm拓撲）代碼如下：
到此為止，所有的Storm元件已經開發完畢！現在，我們把上述工程打成jar包，放到Storm叢集中運作，具體可以到Nimbus對應的Storm安裝目錄下面的bin目錄，輸入：storm jar + {jar路徑}。
比如我這裡是輸入：storm jar /home/tj/install/SensitiveTopology.jar newlandframework.storm.topology.SensitiveTopology，然後，把疑似垃圾短信使用者的垃圾短信内容檔案放到指定的伺服器下面的目錄（/home/tj/data/591、/home/tj/data/592），最後，打開剛才的Storm UI，觀察任務的啟動執行情況，這裡如下圖所示：
可以看到我們剛才送出的拓撲：SensitiveTopology已經成功送出到Storm叢集裡面了。這個時候，你可以滑鼠點選SensitiveTopology，然後，會打開如下的一個Spouts/Bolts的監控界面，如下圖所示：
我們可以很清楚的看到：Spouts元件（使用者短信采集器）：SensitiveFileReader591、SensitiveFileReader592的線程數executors、任務送出emitted情況。以及Bolts元件：監控短信内容拆解分析器（SensitiveFileAnalyzer）、敏感資訊采集處理（SensitiveBatchBolt）的運作情況，這樣監控起來就非常友善。除此之外，我們還可以到對應的Supervisor伺服器對應的Storm安裝目錄下面的logs目錄，檢視一下worker的工作日志，我們來看下敏感資訊監控過濾的處理情況，截圖如下：
通過SensitiveBatchBolt子產品的監控線程，可以看到，我們目前已經采集到了9個敏感資訊使用者了，我們再來看下，這些包含敏感關鍵字的使用者有沒有入庫MySQL成功呢？
發現入庫的結果也是9個，和日志列印的數量上是一緻的。而且垃圾短信内容sms_content果然都包含了“racketeer”、“Bad”這些敏感關鍵字！完全符合我們的預期。而且，以後檔案處理量上來了，我們可以通過調整設定Spouts/Bolts的并行度，和Worker的數量進行化解。當然，你還可以通過水準擴充叢集的數量來解決這個問題。
Storm在Apache開源項目的網址是：http://storm.apache.org/，有興趣的朋友可以經常關注一下。官網上面有很權威的技術規範說明，以及如何把Storm和消息隊列、HDFS、HBase有效的內建起來。目前在國内，就我個人看法，對Storm分析應用，做得最好的應該算是阿裡巴巴，它在原來Storm的基礎上加以改良，開源出JStorm，有興趣的朋友，同樣可以多關注一下。
借助Storm，我們可以很輕松地開發分布式實時處理應用，上述場景的設計，隻是Storm應用的一個案例。相比傳統的單機伺服器應用而言，叢集化的并行協同計算處理，是雲計算、大資料時代的一個趨勢，也是我今後努力學習的方向。故在此寫下，自己的學習經驗體會，抛磚引玉，有說的不對的地方，還請各位園友批評指正！不吝賜教！