SpringBoot3集成Kafka

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

KSQL语句测试

select*from`test-topic`where`partition`in(0)orderby`date`desclimit5

select*from`test-topic`where`partition`in(0)and msg like'%5%'orderby`date`desclimit3

三、工程搭建

1、工程结构

2、依赖管理

这里关于依赖的管理就比较复杂了，首先

spring-kafka

组件选择与boot框架中spring相同的依赖，即

6.0.10

版本，在

spring-kafka

3、配置文件

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

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

四、基础用法

1、消息生产

模板类

KafkaTemplate

用于执行高级的操作，封装各种消息发送的方法，在该方法中，通过

topic

和

key

以及消息主体，实现消息的生产；

@RestControllerpublicclassProducerWeb{@ResourceprivateKafkaTemplate<String,String> kafkaTemplate;@GetMapping("/send/msg")publicString sendMsg (){try{// 构建消息主体JsonMapper jsonMapper =newJsonMapper();String msgBody = jsonMapper.writeValueAsString(newMqMsg(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和消息列表；

@ComponentpublicclassConsumerListener{privatestaticfinalLogger log =LoggerFactory.getLogger(ConsumerListener.class);@KafkaListener(topics ="boot-kafka-topic")publicvoid 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();}}}

五、参考源码

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

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

标签： kafka linq 分布式

本文转载自: https://blog.csdn.net/cicada_smile/article/details/132356427
版权归原作者 知了一笑 所有，如有侵权，请联系我们删除。