SpringBoot3集成Kafka

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: SpringBoot3集成KafkaKafka是一个开源的分布式事件流平台,常被用于高性能数据管道、流分析、数据集成和关键任务应用,基于Zookeeper协调的处理平台,也是一种消息系统,具有更好的吞吐量、内置分区、复制和容错。

标签: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

1.png

KSQL语句测试

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

2.png

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

3.png

三、工程搭建

1、工程结构

4.png

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();
        }
    }
}

5.png

五、参考源码

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

源码仓库:
https://gitee.com/cicadasmile/butte-spring-parent
相关文章
|
11天前
|
消息中间件 Java Kafka
Springboot集成高低版本kafka
Springboot集成高低版本kafka
|
11天前
|
NoSQL Java Redis
SpringBoot集成Redis解决表单重复提交接口幂等(亲测可用)
SpringBoot集成Redis解决表单重复提交接口幂等(亲测可用)
387 0
|
10天前
|
Java 数据库连接 数据安全/隐私保护
springBoot集成token认证,最全Java面试知识点梳理
springBoot集成token认证,最全Java面试知识点梳理
|
11天前
|
NoSQL Java MongoDB
【MongoDB 专栏】MongoDB 与 Spring Boot 的集成实践
【5月更文挑战第11天】本文介绍了如何将非关系型数据库MongoDB与Spring Boot框架集成,以实现高效灵活的数据管理。Spring Boot简化了Spring应用的构建和部署,MongoDB则以其对灵活数据结构的处理能力受到青睐。集成步骤包括:添加MongoDB依赖、配置连接信息、创建数据访问对象(DAO)以及进行数据操作。通过这种方式,开发者可以充分利用两者优势,应对各种数据需求。在实际应用中,结合微服务架构等技术,可以构建高性能、可扩展的系统。掌握MongoDB与Spring Boot集成对于提升开发效率和项目质量至关重要,未来有望在更多领域得到广泛应用。
【MongoDB 专栏】MongoDB 与 Spring Boot 的集成实践
|
11天前
|
消息中间件 JSON Java
RabbitMQ的springboot项目集成使用-01
RabbitMQ的springboot项目集成使用-01
|
11天前
|
搜索推荐 Java 数据库
springboot集成ElasticSearch的具体操作(系统全文检索)
springboot集成ElasticSearch的具体操作(系统全文检索)
|
11天前
|
网络协议 Java Spring
Springboot 集成websocket
Springboot 集成websocket
18 0
|
11天前
|
安全 Java 数据库连接
在IntelliJ IDEA中通过Spring Boot集成达梦数据库:从入门到精通
在IntelliJ IDEA中通过Spring Boot集成达梦数据库:从入门到精通
|
11天前
|
缓存 NoSQL Java
springboot业务开发--springboot集成redis解决缓存雪崩穿透问题
该文介绍了缓存使用中可能出现的三个问题及解决方案:缓存穿透、缓存击穿和缓存雪崩。为防止缓存穿透,可校验请求数据并缓存空值;缓存击穿可采用限流、热点数据预加载或加锁策略;缓存雪崩则需避免同一时间大量缓存失效,可设置随机过期时间。文章还提及了Spring Boot中Redis缓存的配置,包括缓存null值、使用前缀和自定义过期时间,并提供了改造代码以实现缓存到期时间的个性化设置。
|
11天前
|
Java Docker 容器
SpringBoot项目集成XXL-job
SpringBoot项目集成XXL-job
10 1

热门文章

最新文章