关键词:
前言
kafka消息中间键也是我们经常要用到的功能,尤其是在大数据、高并发的项目中,如日志收集、业务数据分发等等。其最核心的俩大功能作用是:①削峰填谷②异步解耦。本节我们主要介绍一下如何在springboot项目中集成kafka消息中间键,实现简单的数据分发以及消费的案例。
正文
-
kafka集群搭建
快速搭建一个kafka集群,我们这里以docker环境为例,搭建一个kafka集群,具体搭建过程请参考以下博客,这里不再赘述。
-
springboot项目引入kafka依赖
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
-
引入kafka配置
spring: #kafka配置 kafka: #kafka集群地址 bootstrap-servers: 192.168.56.10:9091,192.168.56.10:9092,192.168.56.10:9093 producer: #批量发送的数据量大小 batch-size: 16384 #可用发送数量的最大缓存 buffer-memory: 33554432 #key序列化器 key-serializer: org.apache.kafka.common.serialization.StringSerializer #value序列化器 value-serializer: org.apache.kafka.common.serialization.StringSerializer properties: linger.ms: 1 consumer: enable-auto-commit: false auto-commit-interval: 100ms key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer properties: session.timeout.ms: 15000 group-id: atp-group-id
-
创建一个kafka生产者生产消息
①创建一个kafka生产者类KafkaProducer生产消息
package com.yundi.atp.platform.kafka; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; /** * @Author: yanp * @Description: kafka生产者 * @Date: 2021/11/3 18:31 * @Version: 1.0.0 */ @Component @Slf4j public class KafkaProducer @Autowired private KafkaTemplate<Integer, String> kafkaTemplate; public void sendMsg(String topic, String msg) log.info("开始发送kafka消息:[topic:,msg:]", topic, msg); ListenableFuture<SendResult<Integer, String>> sendMsg = kafkaTemplate.send(topic, msg); //消息确认 sendMsg.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() @Override public void onFailure(Throwable throwable) log.error("send error:[ex:,topic:,msg:]", throwable, topic, msg); @Override public void onSuccess(SendResult<Integer, String> stringStringSendResult) log.info("send success:[topic:,msg:]", topic, msg); );
②通过web服务发送生产消息测试
package com.yundi.atp.platform.module.test.controller; import com.yundi.atp.platform.common.Result; import com.yundi.atp.platform.kafka.KafkaProducer; import io.swagger.annotations.Api; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * @Author: yanp * @Description: * @Date: 2021/11/4 17:32 * @Version: 1.0.0 */ @Api(tags = "Springboot集成Kafka测试") @RestController @RequestMapping(value = "/kafka") public class KafkaContoller @Autowired private KafkaProducer kafkaProducer; @GetMapping(value = "/sendKafkaTestMsg/topic/msg") public Result sendKafkaTestMsg(@PathVariable(value = "topic")String topic,@PathVariable(value = "msg")String msg) kafkaProducer.sendMsg(topic,msg); return Result.success();
③通过测试页面发送生产者消息
<template> <div class="container"> <div class="title"> <span>Springboot集成kafka案例</span> <el-divider direction="vertical"></el-divider> <router-link to="home"> <span style="font-size: 18px;">退出</span> </router-link> </div> <el-divider>Test Staring</el-divider> <div style="width: 400px;background: #ddd;padding: 40px 20px;"> <el-form ref="form" :model="form" label-width="70px" class="login"> <el-form-item label="发送主题" prop="topic"> <el-input v-model="form.topic"></el-input> </el-form-item> <el-form-item label="发送消息" prop="name"> <el-input v-model="form.msg"></el-input> </el-form-item> <el-button type="primary" @click="sengMsg" style="width: 100%;margin: 0;">立即发送</el-button> </el-form> </div> </div> </template> <script> export default name: "Kafka", data() return form: topic: 'atp', msg: 'hello world!', , methods: sengMsg() this.$http.get('/kafka/sendKafkaTestMsg/'+this.form.topic+"/"+this.form.msg,).then(res => if (res.data.code === 1) this.$refs.form.resetFields(); this.$message.success(res.data.msg); else this.$refs.form.resetFields(); this.$message.warning(res.data.msg); ).catch(error => this.$message.error(error); ); </script> <style scoped lang="scss"> .container padding: 10px; </style>
-
创建kafka消费者消费消息
package com.yundi.atp.platform.kafka; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; /** * @Author: yanp * @Description: kafka消费者 * @Date: 2021/11/3 18:32 * @Version: 1.0.0 */ @Component @Slf4j public class KafkaConsumer /** * 消费者:监听test主题,分组:groupId1 */ @KafkaListener(topics = "test", groupId = "groupId1") public void consumer1(ConsumerRecord<Integer, String> record) log.info("consumer1 start consume message:[topic:,msg:]", record.topic(), record.value()); /** * 消费者:监听atp主题,分组:groupId1 */ @KafkaListener(topics = "atp", groupId = "groupId1") public void consumer2(ConsumerRecord<Integer, String> record) log.info("consumer2 start consume message:[topic:,msg:]", record.topic(), record.value()); /** * 消费者:监听atp主题,分组:groupId3 */ @KafkaListener(topics = "atp", groupId = "groupId2") public void consumer3(ConsumerRecord<Integer, String> record) log.info("consumer3 start consume message:[topic:,msg:]", record.topic(), record.value());
-
结果测试
①发送生产者消息
②消费者消费消息
结语
ok,到这里关于springboot集成kafka案例实战就结束了,我们下期见。。。
(十八)atp应用测试平台——关于springboot应用监控的那些事(代码片段)
...,快快告诉我你的绝招。本节内容我们主要介绍一下springboot应用的常见应用参数监控指标,从而更好的关注springboot应用的运行状况并实现应用的监控。除此之外,我们通过集成一 查看详情
atp应用测试平台——使用bat批处理实现springboot项目的启动与关闭(代码片段)
前言在windows环境中实现springboot项目的启停,我们可以通过cmd控制台来实现,更为优雅的方式是我们通过批处理文件来处理,这样可以屏蔽项目启动关停的难度,实现一键式的启停,并且自行配置java环境,... 查看详情
(十七)atp应用测试平台——自定义实现一个springboot2的线程池启动器starter
前言启动器是springboot的一大特点,我们可以根据项目自身需求按需装配我们的组件。例如我们需要操作redis,项目中可以添加一个redis的启动器spring-boot-starter-data-redis,这样redis的一些客户端操作功能我们就集成好了... 查看详情
(十七)atp应用测试平台——自定义实现一个springboot2的线程池启动器starter(代码片段)
前言启动器是springboot的一大特点,我们可以根据项目自身需求按需装配我们的组件。例如我们需要操作redis,项目中可以添加一个redis的启动器spring-boot-starter-data-redis,这样redis的一些客户端操作功能我们就集成好了... 查看详情
(十九)atp应用测试平台——springboot集成rocketmq案例实战(代码片段)
...容是关于RocketMQ消息中间键的实战内容,主要介绍在springboot项目中如何集成使用RocketMQ消息中间键,包括消息的发送、消息的接收以及RocketMQ的一些配置说明,以及效果说明。话不多说,开始实战内容。正文搭建Roc... 查看详情
(十四)atp应用测试平台——使用docker-compose一键式安装atp应用测试平台的依赖服务(代码片段)
前言关于ATP应用服务测试平台的相关内容已经更新不少,下载项目的小伙伴第一时间一定是想着怎么把这个平台项目跑起来,看下小编花里胡哨的效果是否能正常show。不过由于依赖的增多,项目的服务也随之多了起... 查看详情
(二十四)atp应用测试平台——springboot集成fastdfs上传与下载功能(代码片段)
前言本节内容我们主要介绍一下如何在springboot项目中集成fastdfs组件,实现文件的上传与下载。关于fastdfs服务中间键的安装过程,本节内容不做介绍。fastdfs是一个轻量级的分布式文件系统,也是我们文件存储中常常... 查看详情
atp应用测试平台——关于axios的配置使用(代码片段)
前言该篇是应某个粉丝的要求讲解一下关于axios请求组件的使用,其实axios组件类似我们以前使用过的jQuery中的ajax组件,都是用于进行http网络请求的组件。axios组件在前后端分离项目中使用的更加广泛,也更易集成... 查看详情
atp应用测试平台——关于网页表格的打印及pdf下载的实战案例(代码片段)
前言在网站应用中,我们可能会有这样一个需求,将网页的部分内容,例如表格,网页片段下载打印或者导出PDF,本小节内容正是关于这样一个内容的实战,基于vue2环境开发,希望能够帮助到你。源码... 查看详情
atp应用测试平台——关于vue-router前端路由的配置使用案例(代码片段)
前言VueRouter是Vue.js(opensnewwindow)官方的路由管理器。它和Vue.js的核心功能深度集成,功能丰富,是我们构建vue前端项目中必不可少的的组件之一,本节我们简单介绍一下在vue项目中如何使用vue-router组件实现页面的路由... 查看详情
atp应用测试平台——关于vue中vue-quill-editormavon-editortinymce等多种富文本编辑器的集成使用(代码片段)
...or以及tinymce等的集成及使用。源代码依然托管在我们的ATP应用测试平台中,源码地址:https://gitee.com/northcangap/atp,仅供参考使用。富文本编辑器实现效果如下: 查看详情
atp应用测试平台——使用easyexcel实现excel导入导出多sheet填充模板下载等功能案例实战(代码片段)
前言Java开发中实现Excel的导入、导出、填充、多sheet页操作等常用功能也是我们经常要面对的开发需求,本文以easyexcel为例,将excel中的常用功能整理成一个个小案例,参考使用。案例源码地址:https://gitee.com/northc... 查看详情
(二十)atp应用测试平台——websocket实现微服务版在线客服聊天室实战案例(代码片段)
前言在前面的博客内容中我们介绍了如何使用websocket实现一个网页版的在线客服聊天室,众所周知,由于websocket是一个长连接,要和服务端保持会话连接,所以其本身并不适用于微服务环境,在微服务环境中... 查看详情
(十七)atp应用测试平台——redis实现api接口访问限流(固定窗口限流算法)(代码片段)
前言开始正文之前,大多数情况下应该有这样一段场景。面试官:说说平常在项目中,你是如何使用redis的?我:我们就很简单啦,比如前后端分离token的存储、短信验证码的存储,权限列表的存储... 查看详情
(二十一)atp应用测试平台——vue实战之大红灯笼高高挂(代码片段)
前言2022年的最后一天班,好的开始,好的结束。把大红灯笼高高挂起来,欢度元旦的到来,兔年的到来。明年再战。明年再见。。。 正文正菜奉上lantern.vue灯笼源码<template><divclass="app"><divclass... 查看详情
(二十一)atp应用测试平台——vue实战之大红灯笼高高挂(代码片段)
前言2022年的最后一天班,好的开始,好的结束。把大红灯笼高高挂起来,欢度元旦的到来,兔年的到来。明年再战。明年再见。。。 正文正菜奉上lantern.vue灯笼源码<template><divclass="app"><divclass... 查看详情
atp应用测试平台——使用vue-video-player视频播放组件实现网页视频流的播放案例实战(代码片段)
前言在网页中播放视频也是我们经常要使用到的功能,例如设备监控的视频流实时播放,MP4、m3u8等视频资源播放等等,在vue项目中,我们可以使用目前封装好的开源组件vue-video-player实现上述的要求。本节我们就... 查看详情
(二十二)atp应用测试平台——swagger2集成swagger-bootstrap-ui实现api文档访问(代码片段)
前言原生的swagger文档虽然也可以作为API文档输出,但是不够简洁优美,本节内容我们介绍一下swagger-bootstrap-ui组件,实现对swaggerAPI文档的二次封装,使用API文档更加简洁明了,无论是入参还是结果响应,... 查看详情