SpringBoot3集成Kafka

这篇具有很好参考价值的文章主要介绍了SpringBoot3集成Kafka。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录
  • 一、简介
  • 二、环境搭建
    • 1、Kafka部署
    • 2、Kafka测试
    • 3、可视化工具
  • 三、工程搭建
    • 1、工程结构
    • 2、依赖管理
    • 3、配置文件
  • 四、基础用法
    • 1、消息生产
    • 2、消息消费
  • 五、参考源码

标签:Kafka3.Kafka-eagle3;

一、简介

Kafka是一个开源的分布式事件流平台,常被用于高性能数据管道、流分析、数据集成和关键任务应用,基于Zookeeper协调的处理平台,也是一种消息系统,具有更好的吞吐量、内置分区、复制和容错,这使得它成为大规模消息处理应用程序的一个很好的解决方案;

二、环境搭建

1、Kafka部署

1、下载安装包:kafka_2.13-3.5.0.tgz

2、配置环境变量

open -e ~/.bash_profile

export KAFKA_HOME=/本地路径/kafka3.5
export PATH=$PATH:$KAFKA_HOME/bin

source ~/.bash_profile

3、该目录【kafka3.5/bin】启动zookeeper
zookeeper-server-start.sh ../config/zookeeper.properties

4、该目录【kafka3.5/bin】启动kafka
kafka-server-start.sh ../config/server.properties

2、Kafka测试

1、生产者
kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
>id-1-message
>id-2-message

2、消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
id-1-message
id-2-message

3、查看topic列表
kafka-topics.sh --bootstrap-server localhost:9092 --list
test-topic

4、查看消息列表
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --partition 0
id-1-message
id-2-message

3、可视化工具

配置和部署

1、下载安装包:kafka-eagle-bin-3.0.2.tar.gz

2、配置环境变量

open -e ~/.bash_profile

export KE_HOME=/本地路径/efak-web-3.0.2
export PATH=$PATH:$KE_HOME/bin

source ~/.bash_profile

3、修改配置文件:system-config.properties

efak.zk.cluster.alias=cluster1
cluster1.zk.list=localhost:2181
efak.url=jdbc:mysql://127.0.0.1:3306/kafka-eagle

4、本地新建数据库:kafka-eagle,注意用户名和密码是否一致

5、启动命令
efak-web-3.0.2/bin/ke.sh start
命令语法: ./ke.sh {start|stop|restart|status|stats|find|gc|jdk|version|sdate|cluster}

6、本地访问【localhost:8048】 username:admin password:123456

SpringBoot3集成Kafka

KSQL语句测试

select * from `test-topic` where `partition` in (0)  order by `date` desc limit 5

SpringBoot3集成Kafka

select * from `test-topic` where `partition` in (0) and msg like '%5%' order by `date` desc limit 3

SpringBoot3集成Kafka

三、工程搭建

1、工程结构

SpringBoot3集成Kafka

2、依赖管理

这里关于依赖的管理就比较复杂了,首先spring-kafka组件选择与boot框架中spring相同的依赖,即6.0.10版本,在spring-kafka最近的版本中3.0.8符合;

但是该版本使用的是kafka-clients组件的3.3.2版本,在Spring文档的kafka模块中,明确说明spring-boot:3.1要使用kafka-clients:3.4,所以从spring-kafka组件中排除掉,重新依赖kafka-clients组件;

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

3、配置文件

配置kafka连接地址,监听器的消息应答机制,消费者的基础模式;

spring:
  # kafka配置
  kafka:
    bootstrap-servers: localhost:9092
    listener:
      missing-topics-fatal: false
      ack-mode: manual_immediate
    consumer:
      group-id: boot-kafka-group
      enable-auto-commit: false
      max-poll-records: 10
      properties:
        max.poll.interval.ms: 3600000

四、基础用法

1、消息生产

模板类KafkaTemplate用于执行高级的操作,封装各种消息发送的方法,在该方法中,通过topickey以及消息主体,实现消息的生产;

@RestController
public class ProducerWeb {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send/msg")
    public String sendMsg (){
        try {
            // 构建消息主体
            JsonMapper jsonMapper = new JsonMapper();
            String msgBody = jsonMapper.writeValueAsString(new MqMsg(7,"boot-kafka-msg"));
            // 发送消息
            kafkaTemplate.send("boot-kafka-topic","boot-kafka-key",msgBody);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        return "OK" ;
    }
}

2、消息消费

编写消息监听类,通过KafkaListener注解控制监听的具体信息,在实现消息生产和消费的方法测试后,使用可视化工具kafka-eagle查看topic和消息列表;

@Component
public class ConsumerListener {

    private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);

    @KafkaListener(topics = "boot-kafka-topic")
    public void listenUser (ConsumerRecord<?,String> record, Acknowledgment acknowledgment) {
        try {
            String key =  String.valueOf(record.key());
            String body = record.value();
            log.info("\n=====\ntopic:boot-kafka-topic,key{},body:{}\n=====\n",key,body);
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            acknowledgment.acknowledge();
        }
    }
}

SpringBoot3集成Kafka文章来源地址https://www.toymoban.com/news/detail-654955.html

五、参考源码

文档仓库:
https://gitee.com/cicadasmile/butte-java-note

源码仓库:
https://gitee.com/cicadasmile/butte-spring-parent

到了这里,关于SpringBoot3集成Kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • SpringBoot3集成Quartz

    目录 一、简介 二、工程搭建 1、工程结构 2、依赖管理 3、数据库 4、配置文件 三、Quartz用法 1、初始化加载 2、新增任务 3、更新任务 4、暂停任务 5、恢复任务 6、执行一次 7、删除任务 8、任务执行 四、参考源码 标签:Quartz.Job.Scheduler; Quartz由Java编写的功能丰富的开源作业

    2024年02月13日
    浏览(38)
  • SpringBoot3集成Zookeeper

    标签:Zookeeper3.8 ,Curator5.5; ZooKeeper是一个集中的服务,用于维护配置信息、命名、提供分布式同步、提供组服务。分布式应用程序以某种形式使用所有这些类型的服务。 1、修改配置文件 2、服务启动 3、客户端测几个增删查的命令 Curator是一组Java库,它让ZooKeeper的使用变得

    2024年01月23日
    浏览(44)
  • SpringBoot3集成RocketMq

    标签:RocketMq5.Dashboard; RocketMQ因其架构简单、业务功能丰富、具备极强可扩展性等特点被广泛应用,比如金融业务、互联网、大数据、物联网等领域的业务场景; 在 rocketmq-starter 组件中,实际上依赖的是 rocketmq-client 组件的 5.0 版本,由于两个新版框架间的兼容问题,需要添

    2024年02月12日
    浏览(47)
  • SpringBoot3集成ElasticSearch

    目录 一、简介 二、环境搭建 1、下载安装包 2、服务启动 三、工程搭建 1、工程结构 2、依赖管理 3、配置文件 四、基础用法 1、实体类 2、初始化索引 3、仓储接口 4、查询语法 五、参考源码 标签:ElasticSearch8.Kibana8; Elasticsearch是一个分布式、RESTful风格的搜索和数据分析引擎

    2024年02月12日
    浏览(39)
  • SpringBoot3数据库集成

    标签:Jdbc.Druid.Mybatis.Plus; 项目工程中,集成数据库实现对数据的增晒改查管理,是最基础的能力,而对于这个功能的实现,其组件选型也非常丰富; 通过如下几个组件来实现数据库的整合; Druid连接池 :阿里开源的数据库连接池,并且提供 SQL 执行的监控能力; MybatisPlu

    2024年02月13日
    浏览(43)
  • Elasticsearch 搜索测试与集成Springboot3

    Elasticsearch是专门做 搜索 的,它非常擅长以下方面的问题 Elasticsearch对模糊搜索非常擅长(搜索速度很快) 从Elasticsearch搜索到的数据可以根据 评分 过滤掉大部分的,只要返回评分高的给用户就好了(原生就支持排序) 没有那么准确的也能搜出相关的结果(能匹配有相

    2024年01月22日
    浏览(44)
  • Java21 + SpringBoot3集成WebSocket

    近日心血来潮想做一个开源项目,目标是做一款可以适配多端、功能完备的模板工程,包含后台管理系统和前台系统,开发者基于此项目进行裁剪和扩展来完成自己的功能开发。 本项目为前后端分离开发,后端基于 Java21 和 SpringBoot3 开发,前端提供了vue、angular、react、uniap

    2024年01月23日
    浏览(53)
  • 【springboot3.x 记录】解决 springboot3 集成 mybatis-plus 报 sqlSession 异常

    2022-12-30,作者最新发布了 3.5.3.1 版本,不需要使用快照版本了 ========================= springboot3 已经发布正式版,第一时间尝鲜看看如何,但是在集成 mybatis-plus 最新版 3.5.2 的时候发现提示异常。 看来 springboot3 在注入这块做了调整,但目前 mybatis-plus 并没有适配到。 于是翻查

    2024年02月13日
    浏览(45)
  • springboot3 集成mybatis 和通用mapper

    xml版本查看:https://www.cnblogs.com/binz/p/6564490.html springboot3.x以前的版本查看 https://www.cnblogs.com/binz/p/17421063.html springboot3.x查看  https://www.cnblogs.com/binz/p/17654403.html 1、pom引用 !-- openapi、 swagger3、knife4j配置,适用boot3 -- !-- https://doc.xiaominfo.com -- dependency groupId com.github.xiaoymin/ groupI

    2024年02月11日
    浏览(44)
  • 【SpringBoot3】Spring Boot 3.0 集成 Redis 缓存

    Redis缓存是一个开源的使用ANSIC语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库,并提供多种语言的API。它主要用于作为数据库、缓存和消息中间件,以快速读写和丰富的数据结构支持而著称。 在应用程序和数据库之间,Redis缓存作为一个中间层起着关键

    2024年02月21日
    浏览(52)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包