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的方法,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对我们网站的支持!

(0)

相关推荐

  • SpringBoot整合Shiro实现登录认证的方法

    安全无处不在,趁着放假读了一下 Shiro 文档,并记录一下 Shiro 整合 Spring Boot 在数据库中根据角色控制访问权限 简介 Apache Shiro是一个功能强大.灵活的,开源的安全框架.它可以干净利落地处理身份验证.授权.企业会话管理和加密. 上图是 Shiro 的基本架构 Authentication(认证) 有时被称为"登录",用来证明用户是用户他们自己本人 Authorization(授权) 访问控制的过程,即确定"谁"访问"什么

  • spring boot整合Shiro实现单点登录的示例代码

    Shiro是什么 Shiro是一个Java平台的开源权限框架,用于认证和访问授权.具体来说,满足对如下元素的支持: 用户,角色,权限(仅仅是操作权限,数据权限必须与业务需求紧密结合),资源(url). 用户分配角色,角色定义权限. 访问授权时支持角色或者权限,并且支持多级的权限定义. Q:对组的支持? A:shiro默认不支持对组设置权限. Q:是否可以满足对组进行角色分配的需求? A:扩展Realm,可以支持对组进行分配角色,其实就是给该组下的所有用户分配权限. Q:对数据权限的支持? 在业务

  • SpringBoot整合Elasticsearch并实现CRUD操作

     配置准备 在build.gradle文件中添加如下依赖: compile "org.elasticsearch.client:transport:5.5.2" compile "org.elasticsearch:elasticsearch:5.5.2" //es 5.x的内部使用的 apache log4日志 compile "org.apache.logging.log4j:log4j-core:2.7" compile "org

  • spring boot整合quartz实现多个定时任务的方法

    最近收到了很多封邮件,都是想知道spring boot整合quartz如何实现多个定时任务的,由于本人生产上并没有使用到多个定时任务,这里给个实现的思路. 1.新建两个定时任务,如下: public class ScheduledJob implements Job{ @Override public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println("sched

  • Spring boot集成Kafka+Storm的示例代码

    前言 由于业务需求需要把Strom与kafka整合到spring boot项目里,实现其他服务输出日志至kafka订阅话题,storm实时处理该话题完成数据监控及其他数据统计,但是网上教程较少,今天想写的就是如何整合storm+kafka 到spring boot,顺带说一说我遇到的坑. 使用工具及环境配置 ​ 1. java 版本jdk-1.8 ​ 2. 编译工具使用IDEA-2017 ​ 3. maven作为项目管理 ​ 4.spring boot-1.5.8.RELEASE 需求体现 1.

  • spring boot整合CAS Client实现单点登陆验证的示例

    本文介绍了spring boot整合CAS Client实现单点登陆验证的示例,分享给大家,也给自己留个笔记,具体如下: 单点登录( Single Sign-On , 简称 SSO )是目前比较流行的服务于企业业务整合的解决方案之一, SSO 使得在多个应用系统中,用户只需要 登录一次 就可以访问所有相互信任的应用系统. CAS Client 负责处理对客户端受保护资源的访问请求,需要对请求方进行身份认证时,重定向到 CAS Server 进行认证.(原则上,客户端应用不再接受任何的用户名密码等

  • spring boot整合mybatis利用Mysql实现主键UUID的方法

    前言 本文主要给大家介绍了关于spring boot整合mybatis利用Mysql实现主键UUID的相关内容,分享出来供大家参考学习,下面话不多说了,来一起看看详细的介绍吧. 实现 基础项目的pom.xml部分代码如下 <properties> <java.version>1.8</java.version> </properties> <!-- Inherit defaults from Spring Boot --> <parent&

  • 浅谈Springboot整合RocketMQ使用心得

    一.阿里云官网---帮助文档 https://help.aliyun.com/document_detail/29536.html?spm=5176.doc29535.6.555.WWTIUh 按照官网步骤,创建Topic.申请发布(生产者).申请订阅(消费者) 二.代码 1.配置: public class MqConfig { /** * 启动测试之前请替换如下 XXX 为您的配置 */ public static final String PUBLIC_TOPIC = "test"

  • Storm框架整合springboot的方法

    Storm:最火的流式处理框架 伴随着信息科技日新月异的发展,信息呈现出爆发式的膨胀,人们获取信息的途径也更加多样.更加便捷,同时对于信息的时效性要求也越来越高.举个搜索场景中的例子,当一个卖家发布了一条宝贝信息时,他希望的当然是这个宝贝马上就可以被卖家搜索出来.点击.购买啦,相反,如果这个宝贝要等到第二天或者更久才可以被搜出来,估计这个大哥就要骂娘了.再举一个推荐的例子,如果用户昨天在淘宝上买了一双袜子,今天想买一副泳镜去游泳,但是却发现系统在不遗余力地给他推荐袜子.鞋子,根本对他今天寻找泳镜

  • JWT整合Springboot的方法步骤

    目录 1.基于JWT认证 1.1 认证流程 1.2 JWT优势 1.3 JWT的结构是什么 2.使用JWT 3.整合Springboot 1.基于JWT认证 1.1 认证流程 首先,前端通过Web表单将自己的用户名和密码发送到后端的接口.这一过程一般是一个HTTP POST请求.建议的方式是通过SSL加密的传输(https协议),从而避免敏感信息被嗅探. 后端核对用户名和密码成功后,将用户的id等其他信息作为JWT Payload(负载),将其与头部分别进行Base64编码拼接后签名,形成一个J

  • activemq整合springboot使用方法(个人微信小程序用)

    主题 ActiveMQ Spring Boot 小程序开发 1.引入依赖 <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.3.RELEASE</version> <relativePath /> <!-- lookup

  • SpringBoot中整合MyBatis-Plus的方法示例

    MyBatis 框架相信大家都用过,虽然 MyBatis 可以直接在 xml 中通过 SQL 语句操作数据库,很是灵活.但正其操作都要通过 SQL 语句进行,就必须写大量的 xml 文件,很是麻烦.于是 MyBatis-Plus 应运而生,作为 MyBatis 的增强工具,更是为我们开发效率得到了质的飞跃. 一.简介 1.MyBatis MyBatis 是一款优秀的持久层框架,它支持自定义 SQL.存储过程以及高级映射.MyBatis 免除了几乎所有的 JDBC 代码以及设置参数和获取结果集的工

  • Springboot安全框架整合SpringSecurity实现方式

    目录 1.工业级安全框架介绍 2.建议搭建Spring Security环境 2.1在pom.xml中添加相关依赖 2.2创建Handler类 2.3创建简单的html和配置相关thymeleaf的路径 2.4最后再加个启动类,那么我们的整合测试就完成勒 2.5成果展示 用户名默认user,密码则随机生成的这串数字 3.进阶版使用 3.1用户名和密码自定义 3.2在config包下创建Encoder进行密码的校验和转码操作 3.3赋予账号角色权限 1.工业级安全框架介绍 Spring Secur

  • SpringBoot框架整合Mybatis简单攻略

    目录 步骤 1 添加mybatis-starter依赖 步骤 2 如何配置mybatis到SpringBoot项目 步骤 3 测试查询 步骤 4 mybatis注解方式 步骤 5 用注解方式做一个新增操作 步骤 6 整合PageHelper分页插件 步骤 7 拓展知识:mybatis四种传参方式 步骤 8 Mybatis中#{}和${}的区别是什么? 步骤 9 Mybatis中模糊查询like语句该怎么写? 步骤 10 SpringBoot整合Mybatis-plus 步骤 11 Mybatis

  • SpringBoot整合Shiro的方法详解

    目录 1.Shito简介 1.1 什么是shiro 1.2 有哪些功能 2.QuickStart 3.SpringBoot中集成 1.导入shiro相关依赖 2.自定义UserRealm 3.定义shiroConfig 4.新建页面进行测试 1.Shito简介 1.1 什么是shiro Apache Shiro是一个java安全(权限)框架 Shiro可以非常容易的开发出足够好的应用,其不仅可以用在javase环境,也可以用在javaee环境 shiro可以完成,认证,授权,加密,会话管理,we

  • SpringBoot详解整合Redis缓存方法

    目录 1.Spring Boot支持的缓存组件 2.基于注解的Redis缓存实现 3.基于API的Redis缓存实现 1.Spring Boot支持的缓存组件 在Spring Boot中,数据的缓存管理存储依赖于Spring框架中cache相关的org.springframework.cache.Cache和org.springframework.cache.CacheManager缓存管理器接口. 如果程序中没有定义类型为CacheManager的Bean组件或者是名为cacheResolve

  • SpringBoot Web开发之系统任务启动与路径映射和框架整合

    目录 本章概要 启动系统任务 CommandLineRunner ApplicationRunner 整合Servlet与Filter和Listener 路径映射 本章概要 启动系统任务 整合 Servlet.Filter 和 Listener 路径映射 启动系统任务 有一些特殊的任务需要在系统启动时执行,例如配置文件加载,数据库初始化等操作.如果没有使用 Spring Boot ,这些问题可以在 Listener 中解决.Spring Boot 对此提供了两种解决方案 CommandLineR

  • thinkPHP5框架整合plupload实现图片批量上传功能的方法

    本文实例讲述了thinkPHP5框架整合plupload实现图片批量上传功能的方法.分享给大家供大家参考,具体如下: 在官网下载plupload http://http//www.plupload.com 或者点击此处本站下载. 这里我们使用的是pluploadQueue 在HTML页面引入相应的css和js,然后根据示例代码修改为自己的代码 <link rel="stylesheet" href="/assets/plupupload/css/jquery.plupl

随机推荐