教你怎样搞定springboot集成kafka

本文分享自华为云社区《手拉手入门springboot+kafka》，作者：QGS。
安装kafka

启动Kafka本地环境需Java 8+以上
Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处置惩罚消费者在网站中的所有动作流数据。
Kafka启动方式有Zookeeper和Kraft，两种方式只能选择此中一种启动，不能同时使用。
Kafka下载https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz

解压tar -xzf kafka_2.13-3.7.0.tgz

一、Zookeeper启动Kafka（kafka内置zookeeper）
Kafka依靠Zookeeper
1、启动Zookeeper 2、启动Kafka
使用kafka自带Zookeeper启动
./zookeeper-server-start.sh ../config/zookeeper.properties &
./zookeeper-server-stop.sh ../config/zookeeper.properties

./kafka-server-start.sh ../config/server.properties &

./kafka-server-stop.sh ../config/server.properties
二、Zookeeper服务器启动Kafka
Zookeeper服务器安装
https://zookeeper.apache.org/

https://dlcdn.apache.org/zookeeper/zookeeper-3.9.2/apache-zookeeper-3.9.2-bin.tar.gz
tar zxvf apache-zookeeper-3.9.2-bin.tar.gz

配置Zookeeper服务器
cp zoo_sample.cfg zoo.cfg

启动Zookeeper服务器

./zkServer.sh start
修改Zookeeper端口

Zoo.cfg添加内容
admin.serverPort=8099
apache-zookeeper-3.9.2-bin/bin目录下重启Zookeeper

Zookeeper服务器启动kafka
/opt/kafka_2.13-3.7.0/bin目录下
./kafka-server-start.sh ../config/server.properties &

Kafka配置文件server.properties

三、使用KRaft启动Kafka

UUID通用唯一识别码（Universally Unique Identifier）
1、天生Cluster UUID(集群UUID)：./kafka-storage.sh random-uuid

2.格式化kafka日记目录：./kafka-storage.sh format -t 3pMJGNJcT0uLIBsZhbucjQ -c ../config/kraft/server.properties

3.启动kafka：./kafka-server-start.sh ../config/kraft/server.properties &

springboot集成kafka

创建topic时，若不指定topic的分区(partition)数量使，则默认为1个分区(partition)

修改server.properties文件

vim server.properties
listeners=PLAINTEXT://0.0.0.0:9092

advertised.listeners=PLAINTEXT://192.168.68.133:9092

springboot加入依靠kafka

1. <dependency>
2. <groupId>org.springframework.kafka</groupId>
3. <artifactId>spring-kafka</artifactId>
4. </dependency>

复制代码

加入spring-kafka依靠后，springboot主动装配好kafkaTemplate的Bean

application.yml配置连接kafka

1. spring:
2. kafka:
3. bootstrap-servers: 192.168.68.133:9092

复制代码

生产者

发送消息

1. @Resource
2. private KafkaTemplate<String,String> kafkaTemplate;
4. @Test
5. void kafkaSendTest(){
6. kafkaTemplate.send("kafkamsg01","hello kafka");
7. }

复制代码

消费者

吸收消息

1. @Component
2. public class KafkaConsumer {
4. @KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")
5. public void consume(String message){
6. System.out.println("接收到消息："+message);
7. }
9. }

复制代码

若没有配置groupid
Failed to start bean 'org.springframework.kafka.config.internalKafkaListenerEndpointRegistry'; nested exception is java.lang.IllegalStateException: No group.id found in consumer config, container properties, or @KafkaListener annotation; a group.id is required when group management is used.

1. @Component
2. public class KafkaConsumer {
4. @KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")
5. public void consume(String message){
6. System.out.println("接收到消息："+message);
7. }
9. }

复制代码

 
点击关注，第一时间相识华为云新鲜技术~

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。