学习 storm，整合 springboot

背景

旧版程序整合 spring，花费了一些功夫，使用起来较为繁琐，遂整合 springboot，简化一些配置。

项目搭建

一、新建 springboot 项目

使用 idea 的 Spring Initializr 创建一个 springboot 项目，名称为 learn-storm；
按需选择依赖库，我只勾选了 lombok；

二、编写 pom.xml

配置项目基本属性 properties，编写依赖库版本

<properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><skipTests>true</skipTests><storm.version>1.1.0</storm.version><kafka.version>0.10.2.2</kafka.version><druid.version>1.1.18</druid.version><mybatis.version>3.4.4</mybatis.version><mybatis-spring.version>1.3.1</mybatis-spring.version><jedis.version>3.2.0</jedis.version><fastjson.version>1.2.68</fastjson.version><hutool.version>5.3.0</hutool.version></properties>

按需引入 dependency，注意使用库的版本和 storm 中自带的版本冲突，如：我使用 logback 日志框架，则需要排除 storm 中的 log4j，等等。
完整 pom.xml 如下，仅作参考

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.2.6.RELEASE</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>inti.tudan</groupId><artifactId>learn-storm</artifactId><version>1.0.0</version><name>learn-storm-with-springboot</name><description>学习 storm，整合 springboot</description><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><skipTests>true</skipTests><storm.version>1.1.0</storm.version><kafka.version>0.10.2.2</kafka.version><druid.version>1.1.18</druid.version><mybatis.version>3.4.4</mybatis.version><mybatis-spring.version>1.3.1</mybatis-spring.version><jedis.version>3.2.0</jedis.version><fastjson.version>1.2.68</fastjson.version><hutool.version>5.3.0</hutool.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId><exclusions><exclusion><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-to-slf4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-jdbc</artifactId></dependency><dependency><groupId>org.springframework.data</groupId><artifactId>spring-data-commons</artifactId></dependency><dependency><groupId>org.springframework.data</groupId><artifactId>spring-data-redis</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.apache.storm</groupId><artifactId>storm-core</artifactId><version>${storm.version}</version><!-- 由于storm环境中有该jar，所以不用pack到最终的task.jar中 --><!--            <scope>provided</scope>--><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion><exclusion><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-slf4j-impl</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.storm</groupId><artifactId>storm-kafka</artifactId><version>${storm.version}</version><exclusions><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_2.10</artifactId><version>${kafka.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>${kafka.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>${druid.version}</version></dependency><dependency><groupId>org.mybatis</groupId><artifactId>mybatis</artifactId><version>${mybatis.version}</version></dependency><dependency><groupId>org.mybatis</groupId><artifactId>mybatis-spring</artifactId><version>${mybatis-spring.version}</version></dependency><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>${jedis.version}</version><exclusions><exclusion><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId></exclusion><exclusion><groupId>log4j</groupId><artifactId>log4j</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><!-- 解析 yaml 的依赖 --><dependency><groupId>org.yaml</groupId><artifactId>snakeyaml</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

三、整理包结构

参考结构如下：

实现原理

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 = "indi.tudan.learn.storm")
@MapperScan("indi.tudan.learn.storm.core.mapper")
public class SpringStormApplication {/*** 非工程启动入口，所以不用 main 方法* 加上 synchronized 的作用是由于 storm 在启动多个 bolt 线程实例时，如果 Springboot 用到 Apollo 分布式配置，会报 ConcurrentModificationException 错误* 详见：https://github.com/ctripcorp/apollo/issues/1658** @param args 参数*/public synchronized static void run(String... args) {SpringApplication app = new SpringApplication(SpringStormApplication.class);// 我们并不需要 web servlet 功能，所以设置为 WebApplicationType.NONEapp.setWebApplicationType(WebApplicationType.NONE);// 忽略掉 banner 输出app.setBannerMode(Banner.Mode.OFF);// 忽略 Spring 启动信息日志app.setLogStartupInfo(false);app.run(args);}
}

与我们传统的Springboot应用启动入口稍微有点区别，主要禁用了web功能，看下正常的启动方式：

public class LearnStormApplication {public static void main(String... args) {// Spring ApplicationContextSpringStormApplication.run();// 提交拓扑任务SpringBeanUtils.getBean(StormTopologySubmit.class).submit();}}

在spout/bolt中调用了SpringStormApplication.run方法后，我们还需要能够拿到ApplicationContext容器对象，这时候我们还需要实现ApplicationContextAware接口，写个工具类 SpringBeanUtils：

@Component
public class SpringBeanUtils implements ApplicationContextAware {/*** spring 上下文*/private static ApplicationContext applicationContext;/*** 获取 spring 上下文** @return spring 上下文* @date 2019年06月25日 16:05:26*/public static ApplicationContext getApplicationContext() {return applicationContext;}/*** 设置 spring 上下文** @param applicationContext spring 上下文* @throws BeansException 异常* @date 2019年06月25日 16:05:20*/@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {SpringBeanUtils.applicationContext = applicationContext;}/*** 获取已注入的 bean，返回 Object** @param beanName 已注入对象名称* @return 已注入对象* @date 2019年06月25日 16:06:04*/public static Object getBean(String beanName) {return applicationContext.getBean(beanName);}/*** 获取已注入的 bean，返回泛型** @param clazz 已注入对象的类类型对象* @param <T>   泛型* @return 已注入泛型对象* @date 2019年06月25日 16:06:33*/public static <T> T getBean(Class<T> clazz) {return applicationContext.getBean(clazz);}/*** 通过name,以及Clazz返回指定的Bean** @param beanName 已注入对象名称* @param clazz    已注入对象的类类型对象* @param <T>      泛型* @return 已注入泛型对象*/public static <T> T getBean(String beanName, Class<T> clazz) {return applicationContext.getBean(beanName, clazz);}}

通过@Component注解使得Spring在启动时能够扫描到该bean，因为BeanUtils实现了ApplicationContextAware接口，Spring会在启动成功时自动调用BeanUtils.setApplicationContext方法，将ApplicationContext对象保存到工具类的静态变量中，之后我们就可以使用BeanUtils.getBean()去获取Spring容器中的bean了。

源码下载

源码在个人 github 上
https://github.com/tudan110/learn-storm

参考资料

Storm框架：Storm整合springboot