SpringBoot3集成Kafka

￡神魔★判官ぃ 2023-10-14 23:00 181阅读 0赞

一、简介

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

二、环境搭建

1、Kafka部署

  1. bash复制代码1、下载安装包：kafka_2.13-3.5.0.tgz
  3. 2、配置环境变量
  5. open -e ~/.bash_profile
  7. export KAFKA_HOME=/本地路径/kafka3.5
  8. export PATH=$PATH:$KAFKA_HOME/bin
  10. source ~/.bash_profile
  12. 3、该目录【kafka3.5/bin】启动zookeeper
  13. zookeeper-server-start.sh ../config/zookeeper.properties
  15. 4、该目录【kafka3.5/bin】启动kafka
  16. kafka-server-start.sh ../config/server.properties

2、Kafka测试

  1. python复制代码1、生产者
  2. kafka-console-producer.sh --broker-list localhost:9092 --topic test-topic
  3. >id-1-message
  4. >id-2-message
  6. 2、消费者
  7. kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic
  8. id-1-message
  9. id-2-message
  11. 3、查看topic列表
  12. kafka-topics.sh --bootstrap-server localhost:9092 --list
  13. test-topic
  15. 4、查看消息列表
  16. kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning --partition 0
  17. id-1-message
  18. id-2-message

3、可视化工具

配置和部署

  1. bash复制代码1、下载安装包：kafka-eagle-bin-3.0.2.tar.gz
  3. 2、配置环境变量
  5. open -e ~/.bash_profile
  7. export KE_HOME=/本地路径/efak-web-3.0.2
  8. export PATH=$PATH:$KE_HOME/bin
  10. source ~/.bash_profile
  12. 3、修改配置文件：system-config.properties
  14. efak.zk.cluster.alias=cluster1
  15. cluster1.zk.list=localhost:2181
  16. efak.url=jdbc:mysql://127.0.0.1:3306/kafka-eagle
  18. 4、本地新建数据库：kafka-eagle，注意用户名和密码是否一致
  20. 5、启动命令
  21. efak-web-3.0.2/bin/ke.sh start
  22. 命令语法: ./ke.sh {start|stop|restart|status|stats|find|gc|jdk|version|sdate|cluster}
  24. 6、本地访问【localhost:8048】 username:admin password:123456

9ed652ac8a975c153d7d9fae7025b6e0.png

get_code_MzVlZjU4YTllNjY2NDYwMjQyOTMyMzcwNjgzZjNlNDYsMTY5MjQxMTg2MzU2Ng

KSQL语句测试

  1. sql复制代码select * from `test-topic` where `partition` in (0)  order by `date` desc limit 5

d2ed28dc7521686163392de98e42281b.png

get_code_Mzc2MzM5MjY5NjI1YmIwZTllMzJiNmEyZTg0MjI2NzAsMTY5MjQxMTg2MzU2Nw

  1. sql复制代码select * from `test-topic` where `partition` in (0) and msg like '%5%' order by `date` desc limit 3

b8b47128c76849bb686591d7a92e71b7.png

get_code_NjMzMmRhMmJkODMxOWNlOWMyNTJjZTdiMDVkMGYwZDgsMTY5MjQxMTg2MzU2Nw

三、工程搭建

1、工程结构

6a755b65aa777fc413a637128ccce7a3.png

get_code_MjBlYWVlOThiOGVkYzIxZmY1YjBjNmJmYjcwNmE3NmUsMTY5MjQxMTg2MzU2Nw

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组件；

  1. xml复制代码<dependency>
  2.     <groupId>org.springframework.kafka</groupId>
  3.     <artifactId>spring-kafka</artifactId>
  4.     <version>${spring-kafka.version}</version>
  5.     <exclusions>
  6.         <exclusion>
  7.             <groupId>org.apache.kafka</groupId>
  8.             <artifactId>kafka-clients</artifactId>
  9.         </exclusion>
  10.     </exclusions>
  11. </dependency>
  12. <dependency>
  13.     <groupId>org.apache.kafka</groupId>
  14.     <artifactId>kafka-clients</artifactId>
  15.     <version>${kafka-clients.version}</version>
  16. </dependency>

3、配置文件

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

  1. yaml复制代码spring:
  2.   # kafka配置
  3.   kafka:
  4.     bootstrap-servers: localhost:9092
  5.     listener:
  6.       missing-topics-fatal: false
  7.       ack-mode: manual_immediate
  8.     consumer:
  9.       group-id: boot-kafka-group
  10.       enable-auto-commit: false
  11.       max-poll-records: 10
  12.       properties:
  13.         max.poll.interval.ms: 3600000

四、基础用法

1、消息生产

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

  1. java复制代码@RestController
  2. public class ProducerWeb {
  4.     @Resource
  5.     private KafkaTemplate<String, String> kafkaTemplate;
  7.     @GetMapping("/send/msg")
  8.     public String sendMsg (){
  9.         try {
  10.             // 构建消息主体
  11.             JsonMapper jsonMapper = new JsonMapper();
  12.             String msgBody = jsonMapper.writeValueAsString(new MqMsg(7,"boot-kafka-msg"));
  13.             // 发送消息
  14.             kafkaTemplate.send("boot-kafka-topic","boot-kafka-key",msgBody);
  15.         } catch (JsonProcessingException e) {
  16.             e.printStackTrace();
  17.         }
  18.         return "OK" ;
  19.     }
  20. }

2、消息消费

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

  1. java复制代码@Component
  2. public class ConsumerListener {
  4.     private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
  6.     @KafkaListener(topics = "boot-kafka-topic")
  7.     public void listenUser (ConsumerRecord<?,String> record, Acknowledgment acknowledgment) {
  8.         try {
  9.             String key =  String.valueOf(record.key());
  10.             String body = record.value();
  11.             log.info("\n=====\ntopic:boot-kafka-topic，key{}，body:{}\n=====\n",key,body);
  12.         } catch (Exception e){
  13.             e.printStackTrace();
  14.         } finally {
  15.             acknowledgment.acknowledge();
  16.         }
  17.     }
  18. }

c9f7459b8eabd2a8308c0ed4a22a2de4.png

get_code_NjIzMTdmM2IzNWQ0YWI4Y2U4MDc1NzRkODFiNTBlOTAsMTY5MjQxMTg2MzU2Nw

五、参考源码

  1. ruby复制代码文档仓库：
  2. https://gitee.com/cicadasmile/butte-java-note
  4. 源码仓库：
  5. https://gitee.com/cicadasmile/butte-spring-parent

发表评论

表情：
评论列表 （有 0 条评论，181人围观）

还没有评论，来说两句吧...

相关阅读

    相关 SpringBoot3集成Kafka

    一、简介 Kafka是一个开源的分布式事件流平台，常被用于高性能数据管道、流分析、数据集成和关键任务应用，基于Zookeeper协调的处理平台，也是一种消息系统，具有更好

    ￡神魔★判官ぃ￡神魔★判官ぃ/ 0 赞/ 182 阅读