storm和spring的结合

1 软件版本 storm0.10 +spring 4.36，网上的例子很少经过摸索除了sport和bolt无法注入其他类都可以了

2 类似于springmvc的结构，bolt service dao

3 因为bolt是由 nimbus 端实例化然后通过序列化传输到supervisor再反向序列化，所以bolt和sport无法通过spring注入

4 SpringBeanutil 类的加载工具

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;public class SpringBeanutil implements ApplicationContextAware {private static ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext arg0)throws BeansException {applicationContext=arg0;}public static <T> T getBean(String name,Class<T> clazz){if (applicationContext==null){SpringContext.SpringContextInit();}return applicationContext.getBean(name,clazz);}}

5 spring加载SpringContext

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringContext {

private static ApplicationContext applicationContext ;

public static synchronized void SpringContextInit(){
       if (applicationContext==null){
           applicationContext =new ClassPathXmlApplicationContext("spring-demo2.xml");
           // applicationContext =new ClassPathXmlApplicationContext("spring-hbase.xml");
       }


   }

6 在bolt的 prepare 中调用对service进行实例化，需要继承BaseRichBolt

public class LoadBolt extends BaseRichBolt {

   /**
   *
   */
   private static final long serialVersionUID = 6084084850932212408L;
   Logger log = LoggerFactory.getLogger(LoadBolt.class);
   private LoadService service ;
   private OutputCollector collector;
   /**
   * 业务处理
   */
   public void execute(Tuple input) {

       service.execute(input, collector);



   }

//申明了输出tuple的key是data
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
       declarer.declare(new Fields("line"));

   }

@Override
   public void prepare(Map map, TopologyContext context, OutputCollector collector) {
       this.collector=collector;
       service=SpringBeanutil.getBean("loadService", LoadService.class);
   }

}

7 service中就实现自己的业务逻辑，应该就是和springmvc一样不过要记得执行collector.ack(input);

经过测试aop功能，类的注入可以顺利使用，不过我使用fatjar来打包的药注意对spring 配置文件的合并，不然会报错，

还有要注意原先在bolt中使用的一些对象放在service中了，要注意有很多类不支持多线程并发操作

service接口

package com.nari.demo2.service;

import backtype.storm.task.OutputCollector;
import backtype.storm.tuple.Tuple;
import org.slf4j.Logger;
public interface BoltService {

String getName();

   void execute(Tuple input ,OutputCollector collector) ;

   Logger getLog();
}

storm和spring的结合相关推荐

Jstorm+Spring+mybatis整合
在现有的jstorm框架下,有一个需求:jstorm要对接mysql数据库的实时读取数据, 通过bolt处理,可能要调用service层的框架,最后保存到数据库. 在网上寻找了一下,发现storm集成 ...
Storm精华问答 | 如何理解spout/bolt的生命周期？
戳蓝字"CSDN云计算"关注我们哦! Storm是Twitter开源的分布式实时大数据处理框架,被业界称为实时版Hadoop.随着越来越多的场景对Hadoop的MapReduce高 ...
Storm入门（七）可靠性机制代码示例
一.关联代码使用maven,代码如下. pom.xml 参考 http://www.cnblogs.com/hd3013779515/p/6970551.html MessageTopology. ...
jstorm mysql_Jstorm+Spring+mybatis整合
在现有的jstorm框架下,有一个需求:jstorm要对接mysql数据库的实时读取数据, 通过bolt处理,可能要调用service层的框架,最后保存到数据库. 在网上寻找了一下,发现storm集成 ...
100 行代码透彻解析 RPC 原理
欢迎关注方志朋的博客,回复"666"获面试宝典文章来源:https://sourl.cn/HpZHvy 引言本文主要论述的是"RPC 实现原理",那么首先 ...
面试官：你给我手写个RPC框架看看
本文主要论述的是"RPC 实现原理",那么首先明确一个问题什么是 RPC 呢?RPC 是 Remote Procedure Call 的缩写,即,远程过程调用.RPC 是一个计算机 ...
透彻解析 RPC 原理
本文主要论述的是"RPC 实现原理",那么首先明确一个问题什么是 RPC 呢?RPC 是 Remote Procedure Call 的缩写,即,远程过程调用.RPC 是一个计算机 ...
Storm构建分布式实时处理应用初探（转）
最近利用闲暇时间,又重新研读了一下Storm.认真对比了一下Hadoop,前者更擅长的是,实时流式数据处理,后者更擅长的是基于HDFS,通过MapReduce方式的离线数据分析计算.对于Hadoop, ...
spring cloud 日志_微服务架构开发实战：ElasticStack实现日志集中化
常见日志集中化的实现方式有许多现成的可用于实现集中式日志记录的解决方案,它们使用不同的方法.体系结构和技术.理解所需的功能并选择满足需求的正确解决方案非常重要. 日志托运有一些日志托运组件可以与其 ...

storm和spring的结合

storm和spring的结合相关推荐

最新文章

热门文章