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包:

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

<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>

<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相关依赖为

“`java

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter</artifactid>

<exclusions>

<exclusion>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter-logging</artifactid>

</exclusion>

</exclusions>

</dependency>

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter-web</artifactid>

</dependency>

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter-aop</artifactid>

</dependency>

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter-test</artifactid>

<scope>test</scope>

</dependency>

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-starter-log4j2</artifactid>

</dependency>

<dependency>

<groupid>org.mybatis.spring.boot</groupid>

<artifactid>mybatis-spring-boot-starter</artifactid>

<version>${mybatis-spring.version}</version>

</dependency>

<dependency>

<groupid>org.springframework.boot</groupid>

<artifactid>spring-boot-configuration-processor</artifactid>

<optional>true</optional>

</dependency>

ps:maven的jar包仅因为项目使用需求,不是最精简,仅供大家参考.

项目结构:

config-存储不同环境配置文件

存储构建spring boot 相关实现类其他如构建名

启动spring boot的时候我们会发现

其实开始整合前,对storm了解的较少,属于刚开始没有接触过,后面参考发现整合到spring boot里面启动spring boot之后并没有相应的方式去触发提交topolgy的函数,所以也造成了以为启动spring boot之后就完事了结果等了半个小时什么事情都没发生才发现没有实现触发提交函数.

为了解决这个问题我的想法是: 启动spring boot->创建kafka监听topic然后启动topolgy完成启动,可是这样的问题kafka监听这个主题会重复触发topolgy,这明显不是我们想要的.看了一会后发现spring 有相关启动完成之后执行某个时间方法,这个对我来说简直是救星啊.所以现在触发topolgy的思路变为:

启动spring boot ->执行触发方法->完成相应的触发条件

构建方法为: