Storm框架：Storm整合springboot

标签：group   service   over   Collector   syn   startup   过程   pac   ica   

实现原理

Storm框架中的每个Spout和Bolt都相当于独立的应用，Strom在启动spout和bolt时提供了一个open方法(spout)和prepare方法(bolt)。我们可以把初始化Spring应用的操作放在这里，这样可以保证每个spout/bolt应用在后续执行过程中都能获取到Spring的ApplicationContext，有了ApplicationContext实例对象，Spring的所有功能就都能用上了。

Spout.open方法实现br/>@Override
public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
//启动Springboot应用
SpringStormApplication.run();

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

}br/>Bolt.prepare方法实现
@Override
public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
//启动Springboot应用
SpringStormApplication.run();

this.map = map;
this.topologyContext = topologyContext;
this.outputCollector = outputCollector;

}br/>SpringStormApplication启动类
@SpringBootApplication
@ComponentScan(value = "com.xxx.storm")
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.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);br/>}
}
在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);
}

在FilterBolt的execute方法中获取Spring beanbr/>@Override
public void execute(Tuple tuple) {
FilterService filterService = (FilterService) BeanUtils.getBean("filterService");
filterService.deleteAll();
}
定义FilterService类，这时候我们就可以使用Spring的相关注解，自动注入，Spring Jpa等功能了。br/>@Service("filterService")
public class FilterService {br/>@Autowired
UserRepository userRepository;

public void deleteAll() {
    userRepository.deleteAll();
}

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>
OK，完美整合！

Storm框架：Storm整合springboot

标签：group   service   over   Collector   syn   startup   过程   pac   ica   

原文地址：http://blog.51cto.com/14084556/2321808