教你如何搞定 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

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

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

application.yml 配置连接 kafka

spring:kafka:bootstrap-servers: 192.168.68.133:9092

生产者

发送消息

@Resourceprivate KafkaTemplate<String,String> kafkaTemplate;
@Testvoid kafkaSendTest(){kafkaTemplate.send("kafkamsg01","hello kafka");}

消费者

接收消息

@Componentpublic class KafkaConsumer {
@KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")public void consume(String message){System.out.println("接收到消息："+message);}
}

若没有配置 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.

@Componentpublic class KafkaConsumer {
@KafkaListener(topics = {"kafkamsg01","test"},groupId = "123")public void consume(String message){System.out.println("接收到消息："+message);}
}

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