storm：最火的流式處理框架
伴隨著信息科技日新月異的發展，信息呈現出爆發式的膨脹，人們獲取信息的途徑也更加多樣、更加便捷，同時對于信息的時效性要求也越來越高。舉個搜索場景中的例子，當一個賣家發布了一條寶貝信息時，他希望的當然是這個寶貝馬上就可以被賣家搜索出來、點擊、購買啦，相反，如果這個寶貝要等到第二天或者更久才可以被搜出來，估計這個大哥就要罵娘了。再舉一個推薦的例子，如果用戶昨天在淘寶上買了一雙襪子，今天想買一副泳鏡去游泳，但是卻發現系統在不遺余力地給他推薦襪子、鞋子，根本對他今天尋找泳鏡的行為視而不見，估計這哥們心里就會想推薦你妹呀。其實稍微了解點背景知識的碼農們都知道，這是因為后臺系統做的是每天一次的全量處理，而且大多是在夜深人靜之時做的，那么你今天白天做的事情當然要明天才能反映出來啦。
•實現一個實時計算系統
全量數據處理使用的大多是鼎鼎大名的hadoop或者hive，作為一個批處理系統，hadoop以其吞吐量大、自動容錯等優點，在海量數據處理上得到了廣泛的使用。但是，hadoop不擅長實時計算，因為它天然就是為批處理而生的，這也是業界一致的共識。否則最近這兩年也不會有s4,storm,puma這些實時計算系統如雨后春筍般冒出來啦。先拋開s4,storm,puma這些系統不談，我們首先來看一下，如果讓我們自己設計一個實時計算系統，我們要解決哪些問題。
1.低延遲。都說了是實時計算系統了，延遲是一定要低的。
2.高性能。性能不高就是浪費機器，浪費機器是要受批評的哦。
3.分布式。系統都是為應用場景而生的，如果你的應用場景、你的數據和計算單機就能搞定，那么不用考慮這些復雜的問題了。我們所說的是單機搞不定的情況。
4.可擴展。伴隨著業務的發展，我們的數據量、計算量可能會越來越大，所以希望這個系統是可擴展的。
5.容錯。這是分布式系統中通用問題。一個節點掛了不能影響我的應用。
好，如果僅僅需要解決這5個問題，可能會有無數種方案，而且各有千秋，隨便舉一種方案，使用消息隊列+分布在各個機器上的工作進程就ok啦。我們再繼續往下看。
1.容易在上面開發應用程序。親，你設計的系統需要應用程序開發人員考慮各個處理組件的分布、消息的傳遞嗎？如果是，那有點麻煩啊，開發人員可能會用不好，也不會想去用。
2.消息不丟失。用戶發布的一個寶貝消息不能在實時處理的時候給丟了，對吧？更嚴格一點，如果是一個精確數據統計的應用，那么它處理的消息要不多不少才行。這個要求有點高哦。
誕 生
在2011年storm開源之前，由于hadoop的火紅，整個業界都在喋喋不休地談論大數據。hadoop的高吞吐，海量數據處理的能力使得人們可以方便地處理海量數據。但是，hadoop的缺點也和它的優點同樣鮮明——延遲大，響應緩慢，運維復雜。
有需求也就有創造，在hadoop基本奠定了大數據霸主地位的時候，很多的開源項目都是以彌補hadoop的實時性為目標而被創造出來。而在這個節骨眼上storm橫空出世了。
storm帶著流式計算的標簽華麗麗滴出場了，看看它的一些賣點：
•分布式系統：可橫向拓展,現在的項目不帶個分布式特性都不好意思開源。
•運維簡單：storm的部署的確簡單。雖然沒有mongodb的解壓即用那么簡單，但是它也就是多安裝兩個依賴庫而已。
•高度容錯：模塊都是無狀態的，隨時宕機重啟。
•無數據丟失：storm創新性提出的ack消息追蹤框架和復雜的事務性處理,能夠滿足很多級別的數據處理需求。不過，越高的數據處理需求，性能下降越嚴重。
•多語言：實際上，storm的多語言更像是臨時添加上去似的。因為，你的提交部分還是要使用java實現。
下面介紹下storm框架整合springboot的方法
我們知道storm本身是一個獨立運行的分布式流式數據處理框架，springboot也是一個獨立運行的web框架。那么如何在strom框架中集成springboot使得我們能夠在storm開發中運用spring的ioc容器及其他如spring jpa等功能呢？我們先來了解以下概念：
•storm主要的三個component：topology、spout、bolt。topology作為主進程控制著spout、bolt線程的運行，他們相當于獨立運行的容器分布于storm集群中的各個機器節點。
•springapplication：是配置spring應用上下文的起點。通過調用springapplication.run()方法它將創建applicationcontext實例，這是我們能夠使用ioc容器的主要beanfactory。之后spring將會加載所有單例模式的beans，并啟動后臺運行的commandlinerunner beans等。
•applicationcontextaware：這是我們能夠在普通java類中調用spring容器里的beans的關鍵接口。
實現原理
storm框架中的每個spout和bolt都相當于獨立的應用，strom在啟動spout和bolt時提供了一個open方法(spout)和prepare方法(bolt)。我們可以把初始化spring應用的操作放在這里，這樣可以保證每個spout/bolt應用在后續執行過程中都能獲取到spring的applicationcontext，有了applicationcontext實例對象，spring的所有功能就都能用上了。
•spout.open方法實現
|
1
2
3
4
5
6
7
8
9
|
@overridepublic void open(map map, topologycontext topologycontext, spoutoutputcollector spoutoutputcollector) { //啟動springboot應用 springstormapplication.run(); this.map = map; this.topologycontext = topologycontext; this.spoutoutputcollector = spoutoutputcollector;} |
•bolt.prepare方法實現
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
@overridepublic void prepare(map map, topologycontext topologycontext, outputcollector outputcollector) { //啟動springboot應用 springstormapplication.run(); this.map = map; this.topologycontext = topologycontext; this.outputcollector = outputcollector;}•springstormapplication啟動類@springbootapplication@componentscan(value = "com.xxx.storm")public class springstormapplication { /** * 非工程啟動入口，所以不用main方法 * @param args */ public static void run(string ...args) { springapplication app = new springapplication(springstormapplication.class); //我們并不需要web servlet功能，所以設置為webapplicationtype.none app.setwebapplicationtype(webapplicationtype.none); //忽略掉banner輸出 app.setbannermode(banner.mode.off); //忽略spring啟動信息日志 app.setlogstartupinfo(false); app.run(args); }} |
與我們傳統的springboot應用啟動入口稍微有點區別，主要禁用了web功能，看下正常的啟動方式：
|
1
2
3
4
5
6
7
|
@springbootapplication@componentscan(value = "com.xxx.web")public class platformapplication { public static void main(string[] args) { springapplication.run(platformapplication.class, args); }} |
•在spout/bolt中調用了springstormapplication.run方法后，我們還需要能夠拿到applicationcontext容器對象，這時候我們還需要實現applicationcontextaware接口，寫個工具類beanutils：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
@componentpublic class beanutils implements applicationcontextaware { private static applicationcontext applicationcontext = null; @override public void setapplicationcontext(applicationcontext applicationcontext) throws beansexception { if (beanutils.applicationcontext == null) { beanutils.applicationcontext = applicationcontext; } } public static applicationcontext getapplicationcontext() { return applicationcontext; } public static object getbean(string name) { return getapplicationcontext().getbean(name); } public static <t> t getbean(class<t> clazz) { return getapplicationcontext().getbean(clazz); } public static <t> t getbean(string name, class<t> clazz) { return getapplicationcontext().getbean(name, clazz); }} |
通過@component注解使得spring在啟動時能夠掃描到該bean，因為beanutils實現了applicationcontextaware接口，spring會在啟動成功時自動調用beanutils.setapplicationcontext方法，將applicationcontext對象保存到工具類的靜態變量中，之后我們就可以使用beanutils.getbean()去獲取spring容器中的bean了。
寫個簡單例子
•在filterbolt的execute方法中獲取spring bean
|
1
2
3
4
5
|
@overridepublic void execute(tuple tuple) { filterservice filterservice = (filterservice) beanutils.getbean("filterservice"); filterservice.deleteall();} |
•定義filterservice類，這時候我們就可以使用spring的相關注解，自動注入，spring jpa等功能了。
|
1
2
3
4
5
6
7
8
9
|
@service("filterservice")public class filterservice { @autowired userrepository userrepository; public void deleteall() { userrepository.deleteall(); }} |
將storm應用作為springboot工程的一個子模塊
工程主目錄的pom文件還是springboot相關的依賴，在storm子模塊中引入storm依賴，這時候啟動strom的topology應用會有一個日志包依賴沖突。
|
1
2
3
4
5
|
slf4j: class path contains multiple slf4j bindings.slf4j: found binding in [jar:file:/applications/intellij%20idea.app/contents/bin/~/.m2/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.11.1/log4j-slf4j-impl-2.11.1.jar!/org/slf4j/impl/staticloggerbinder.class]slf4j: found binding in [jar:file:/applications/intellij%20idea.app/contents/bin/~/.m2/repository/ch/qos/logback/logback-classic/1.2.3/logback-classic-1.2.3.jar!/org/slf4j/impl/staticloggerbinder.class]slf4j: see http://www.slf4j.org/codes.html#multiple_bindings for an explanation.slf4j: actual binding is of type [org.apache.logging.slf4j.log4jloggerfactory] |
我們需要在storm子模塊的pom文件中重寫org.springframework.boot:spring-boot-starter包依賴，將springboot的相關日志包排除掉，如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter</artifactid> <exclusions> <exclusion> <groupid>org.apache.logging.log4j</groupid> <artifactid>log4j-to-slf4j2</artifactid> </exclusion> <exclusion> <groupid>ch.qos.logback</groupid> <artifactid>logback-classic2</artifactid> </exclusion> </exclusions></dependency> |
總結
以上所述是小編給大家介紹的storm框架整合springboot的方法，希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時回復大家的。在此也非常感謝大家對服務器之家網站的支持！
原文鏈接：https://www.cnblogs.com/gouyg/p/storm-springboot.html
