Storm框架整合springboot的方法

發布時間: 2019-06-18 18:27:59 來源: 互聯網 欄目: Java 點擊:

Storm框架中的每個Spout和Bolt都相當于獨立的應用,Strom在啟動spout和bolt時提供了一個open方法(spout)和prepare方法(bolt)。這篇文章主要介紹了Storm框架整合springboot的方法,需要的朋友可以參考下

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方法實現

@Override
public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
 //啟動Springboot應用
 SpringStormApplication.run();

 this.map = map;
 this.topologyContext = topologyContext;
 this.spoutOutputCollector = spoutOutputCollector;
}

•Bolt.prepare方法實現

@Override
public 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功能,看下正常的啟動方式:

@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:

@Component
public 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

@Override
public void execute(Tuple tuple) {
 FilterService filterService = (FilterService) BeanUtils.getBean("filterService");
 filterService.deleteAll();
}

•定義FilterService類,這時候我們就可以使用Spring的相關注解,自動注入,Spring Jpa等功能了。

@Service("filterService")
public class FilterService {
 @Autowired
 UserRepository userRepository;
 
 public void deleteAll() {
  userRepository.deleteAll();
 }
}

將storm應用作為Springboot工程的一個子模塊

工程主目錄的pom文件還是springboot相關的依賴,在storm子模塊中引入storm依賴,這時候啟動Strom的topology應用會有一個日志包依賴沖突。

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的相關日志包排除掉,如下:

<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的方法,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會及時回復大家的。在此也非常感謝大家對我們網站的支持!

本文標題: Storm框架整合springboot的方法
本文地址: http://www.leskzw.tw/ruanjian/java/244055.html

如果認為本文對您有所幫助請贊助本站

支付寶掃一掃贊助微信掃一掃贊助

  • 支付寶掃一掃贊助
  • 微信掃一掃贊助
  • 支付寶先領紅包再贊助
    聲明:凡注明"本站原創"的所有文字圖片等資料,版權均屬編程客棧所有,歡迎轉載,但務請注明出處。
    Spring Boot2.0 @ConfigurationProperties使用詳解springboot v2.0.3版本多數據源配置方法
    Top 广东好彩1中奖规则