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

为了使用spring boot 统一管理各种微服务,及同时避免多个分散配置

2.具体思路及整合原因

​ 使用spring boot统一管理kafka、storm、redis等所需要的bean,通过其他服务日志收集至Kafka,KafKa实时发送日志至storm,在strom bolt时进行相应的处理操作

遇到的问题

​ 1.使用spring boot并没有相关整合storm

​ 2.以spring boot启动方式不知道如何触发提交Topolgy

​ 3.提交Topology时遇到numbis not client localhost 问题

​ 4.Storm bolt中无法通过注解获得实例化bean进行相应的操作

解决思路

在整合之前我们需要知道相应的spring boot 的启动方式及配置(如果你在阅读本文时，默认你已经对storm,kafka及spring boot有相关了解及使用)

spring boot 对storm进行整合的例子在网上很少，但是因为有相应的需求,因此我们还是需要整合.

首先导入所需要jar包:

<dependency>
 <groupId>org.apache.kafka</groupId>
 <artifactId>kafka-clients</artifactId>
 <version>0.10.1.1</version>
 </dependency>

 <dependency>
 <groupId>org.springframework.cloud</groupId>
 <artifactId>spring-cloud-starter-stream-kafka</artifactId>
 <exclusions>
 <exclusion>
  <artifactId>zookeeper</artifactId>
  <groupId>org.apache.zookeeper</groupId>
 </exclusion>
 <exclusion>
  <artifactId>spring-boot-actuator</artifactId>
  <groupId>org.springframework.boot</groupId>
 </exclusion>
 <exclusion>
  <artifactId>kafka-clients</artifactId>
  <groupId>org.apache.kafka</groupId>
 </exclusion>
 </exclusions>
 </dependency>

 <dependency>
 <groupId>org.springframework.kafka</groupId>
 <artifactId>spring-kafka</artifactId>
 <exclusions>
 <exclusion>
  <artifactId>kafka-clients</artifactId>
  <groupId>org.apache.kafka</groupId>
 </exclusion>
 </exclusions>
 </dependency>

 <dependency>
 <groupId>org.springframework.data</groupId>
 <artifactId>spring-data-hadoop</artifactId>
 <version>2.5.0.RELEASE</version>
 <exclusions>
 <exclusion>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-log4j12</artifactId>
 </exclusion>
 <exclusion>
  <artifactId>commons-logging</artifactId>
  <groupId>commons-logging</groupId>
 </exclusion>
 <exclusion>
  <artifactId>netty</artifactId>
  <groupId>io.netty</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-core-asl</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>curator-client</artifactId>
  <groupId>org.apache.curator</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jettison</artifactId>
  <groupId>org.codehaus.jettison</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-mapper-asl</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-jaxrs</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>snappy-java</artifactId>
  <groupId>org.xerial.snappy</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-xc</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>guava</artifactId>
  <groupId>com.google.guava</groupId>
 </exclusion>
 <exclusion>
  <artifactId>hadoop-mapreduce-client-core</artifactId>
  <groupId>org.apache.hadoop</groupId>
 </exclusion>
 <exclusion>
  <artifactId>zookeeper</artifactId>
  <groupId>org.apache.zookeeper</groupId>
 </exclusion>
 <exclusion>
  <artifactId>servlet-api</artifactId>
  <groupId>javax.servlet</groupId>
 </exclusion>

 </exclusions>
 </dependency>
 <dependency>
 <groupId>org.apache.zookeeper</groupId>
 <artifactId>zookeeper</artifactId>
 <version>3.4.10</version>
 <exclusions>
 <exclusion>
  <artifactId>slf4j-log4j12</artifactId>
  <groupId>org.slf4j</groupId>
 </exclusion>
 </exclusions>
 </dependency>
 <dependency>
 <groupId>org.apache.hbase</groupId>
 <artifactId>hbase-client</artifactId>
 <version>1.2.4</version>
 <exclusions>
 <exclusion>
  <artifactId>log4j</artifactId>
  <groupId>log4j</groupId>
 </exclusion>
 <exclusion>
  <artifactId>zookeeper</artifactId>
  <groupId>org.apache.zookeeper</groupId>
 </exclusion>
 <exclusion>
  <artifactId>netty</artifactId>
  <groupId>io.netty</groupId>
 </exclusion>
 <exclusion>
  <artifactId>hadoop-common</artifactId>
  <groupId>org.apache.hadoop</groupId>
 </exclusion>
 <exclusion>
  <artifactId>guava</artifactId>
  <groupId>com.google.guava</groupId>
 </exclusion>
 <exclusion>
  <artifactId>hadoop-annotations</artifactId>
  <groupId>org.apache.hadoop</groupId>
 </exclusion>
 <exclusion>
  <artifactId>hadoop-yarn-common</artifactId>
  <groupId>org.apache.hadoop</groupId>
 </exclusion>
 <exclusion>
  <artifactId>slf4j-log4j12</artifactId>
  <groupId>org.slf4j</groupId>
 </exclusion>
 </exclusions>
 </dependency>
 <dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-common</artifactId>
 <version>2.7.3</version>
 <exclusions>
 <exclusion>
  <artifactId>commons-logging</artifactId>
  <groupId>commons-logging</groupId>
 </exclusion>
 <exclusion>
  <artifactId>curator-client</artifactId>
  <groupId>org.apache.curator</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-mapper-asl</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>jackson-core-asl</artifactId>
  <groupId>org.codehaus.jackson</groupId>
 </exclusion>
 <exclusion>
  <artifactId>log4j</artifactId>
  <groupId>log4j</groupId>
 </exclusion>
 <exclusion>
  <artifactId>snappy-java</artifactId>
  <groupId>org.xerial.snappy</groupId>
 </exclusion>
 <exclusion>
  <artifactId>zookeeper</artifactId>
  <groupId>org.apache.zookeeper</groupId>
 </exclusion>
 <exclusion>
  <artifactId>guava</artifactId>
  <groupId>com.google.guava</groupId>
 </exclusion>
 <exclusion>
  <artifactId>hadoop-auth</artifactId>
  <groupId>org.apache.hadoop</groupId>
 </exclusion>
 <exclusion>
  <artifactId>commons-lang</artifactId>
  <groupId>commons-lang</groupId>
 </exclusion>
 <exclusion>
  <artifactId>slf4j-log4j12</artifactId>
  <groupId>org.slf4j</groupId>
 </exclusion>
 <exclusion>
  <artifactId>servlet-api</artifactId>
  <groupId>javax.servlet</groupId>
 </exclusion>
 </exclusions>
 </dependency>
 <dependency>
 <groupId>org.apache.hadoop</groupId>
 <artifactId>hadoop-mapreduce-examples</artifactId>
 <version>2.7.3</version>
 <exclusions>
 <exclusion>
  <artifactId>commons-logging</artifactId>
  <groupId>commons-logging</groupId>
 </exclusion>
 <exclusion>
  <artifactId>netty</artifactId>
  <groupId>io.netty</groupId>
 </exclusion>
 <exclusion>
  <artifactId>guava</artifactId>
  <groupId>com.google.guava</groupId>
 </exclusion>
 <exclusion>
  <artifactId>log4j</artifactId>
  <groupId>log4j</groupId>
 </exclusion>
 <exclusion>
  <artifactId>servlet-api</artifactId>
  <groupId>javax.servlet</groupId>
 </exclusion>
 </exclusions>
 </dependency>

 <!--storm-->
 <dependency>
 <groupId>org.apache.storm</groupId>
 <artifactId>storm-core</artifactId>
 <version>${storm.version}</version>
 <scope>${provided.scope}</scope>
 <exclusions>
 <exclusion>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-slf4j-impl</artifactId>
 </exclusion>
 <exclusion>
  <artifactId>servlet-api</artifactId>
  <groupId>javax.servlet</groupId>
 </exclusion>
 </exclusions>
 </dependency>

 <dependency>
 <groupId>org.apache.storm</groupId>
 <artifactId>storm-kafka</artifactId>
 <version>1.1.1</version>
 <exclusions>
 <exclusion>
  <artifactId>kafka-clients</artifactId>
  <groupId>org.apache.kafka</groupId>
 </exclusion>
 </exclusions>
 </dependency>

其中去除jar包是因为需要相与项目构建依赖有多重依赖问题,storm版本为1.1.0  spring boot相关依赖为

（编辑：安卓应用网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!