教你如何搞定 springboot 集成 kafka
![教你如何搞定springboot集成kafka](https://static001.geekbang.org/infoq/39/3999c6ae988189fcaf7dce774a8ca54f.jpeg)
本文分享自华为云社区《手拉手入门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
![](https://static001.geekbang.org/infoq/33/33052937ce28aa1d68c44204603f0059.png)
解压 tar -xzf kafka_2.13-3.7.0.tgz
![](https://static001.geekbang.org/infoq/68/6848c2abae378d1ae5f46dc86d4a66c7.png)
一、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
![](https://static001.geekbang.org/infoq/6c/6c71227008788970d9ba20a9c7969755.png)
./kafka-server-start.sh ../config/server.properties &
![](https://static001.geekbang.org/infoq/6f/6fd24e9378c67dc5601f42f4cad46be1.png)
./kafka-server-stop.sh ../config/server.properties
二、Zookeeper 服务器启动 Kafka
Zookeeper 服务器安装
![](https://static001.geekbang.org/infoq/85/85472cd6c2ca04cd918c1afa6b1ac838.png)
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
![](https://static001.geekbang.org/infoq/b2/b2502c574cb8da49f5bcd3b7de86a6c0.png)
![](https://static001.geekbang.org/infoq/f8/f8cc35d6660c8be3551d75e975119abe.png)
配置 Zookeeper 服务器
cp zoo_sample.cfg zoo.cfg
![](https://static001.geekbang.org/infoq/b2/b23ac67465ed9af95391ba3fa387dabb.png)
启动 Zookeeper 服务器
![](https://static001.geekbang.org/infoq/d3/d33bdc7267c86a969243ff8ce0c77adc.png)
./zkServer.sh start
修改 Zookeeper 端口
![](https://static001.geekbang.org/infoq/d7/d7277b11d879efbcd423e0f32f1eaa5c.png)
Zoo.cfg 添加内容
admin.serverPort=8099
apache-zookeeper-3.9.2-bin/bin 目录下重启 Zookeeper
![](https://static001.geekbang.org/infoq/5c/5cbda5c9f8a49d82edaae61d5272159d.png)
Zookeeper 服务器启动 kafka
/opt/kafka_2.13-3.7.0/bin 目录下
./kafka-server-start.sh ../config/server.properties &
![](https://static001.geekbang.org/infoq/b7/b7316909201d854841078367053aedbf.png)
Kafka 配置文件 server.properties
![](https://static001.geekbang.org/infoq/50/504267c499c286dc4848bec233ea9c08.png)
三、使用 KRaft 启动 Kafka
![](https://static001.geekbang.org/infoq/78/78fe417c09df675633c8dd470f0a5e79.png)
UUID 通用唯一识别码(Universally Unique Identifier)
1、生成 Cluster UUID(集群 UUID):./kafka-storage.sh random-uuid
![](https://static001.geekbang.org/infoq/c6/c6c8ce88534648511ce91d25d9d955b7.png)
2.格式化 kafka 日志目录:./kafka-storage.sh format -t 3pMJGNJcT0uLIBsZhbucjQ -c ../config/kraft/server.properties
![](https://static001.geekbang.org/infoq/b7/b7a9feac8513b85c37e2e7e0c31689be.png)
3.启动 kafka:./kafka-server-start.sh ../config/kraft/server.properties &
![](https://static001.geekbang.org/infoq/03/03901a16b7deed3c49b1f54168b1f7d5.png)
springboot 集成 kafka
![](https://static001.geekbang.org/infoq/78/7800bc26e85f5a0a28036eee3ad40e4c.png)
创建 topic 时,若不指定 topic 的分区(partition)数量使,则默认为 1 个分区(partition)
![](https://static001.geekbang.org/infoq/a4/a4e188c76b316925eb296e9a95943351.png)
修改 server.properties 文件
vim server.properties
listeners=PLAINTEXT://0.0.0.0:9092
![](https://static001.geekbang.org/infoq/5a/5accbac4263efb0c3da3a540a1bc7303.png)
advertised.listeners=PLAINTEXT://192.168.68.133:9092
![](https://static001.geekbang.org/infoq/9d/9d49ee96118ea613ef38cacfcec46741.png)
![](https://static001.geekbang.org/infoq/f2/f23849e2e8107dbc0fa86e747064206e.png)
![](https://static001.geekbang.org/infoq/f6/f67cb7766bff2877cafc1ba22f5345e4.png)
springboot 加入依赖 kafka
加入 spring-kafka 依赖后,springboot 自动装配好 kafkaTemplate 的 Bean
![](https://static001.geekbang.org/infoq/de/dea8330e97bfd407d2d8148965cd55f3.png)
application.yml 配置连接 kafka
生产者
发送消息
![](https://static001.geekbang.org/infoq/fa/fa6d1081055eaa38a835106d4f88cf5c.png)
![](https://static001.geekbang.org/infoq/66/66bd5a8b799ec10cf8787d82ba8a79e3.png)
消费者
接收消息
![](https://static001.geekbang.org/infoq/e5/e5bf542559b6d376bd536ac4b378dce3.png)
若没有配置 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.
![](https://static001.geekbang.org/infoq/69/69c27ee94c2801c0c8a647fd0d5a433f.png)
![](https://static001.geekbang.org/infoq/1f/1fe6435f5a2fbe64bf6fd28d17389c3b.png)
![](https://static001.geekbang.org/infoq/4c/4c8136d375899324b46227868e720cab.png)
版权声明: 本文为 InfoQ 作者【华为云开发者联盟】的原创文章。
原文链接:【http://xie.infoq.cn/article/402c9399d8fe229af9e637451】。文章转载请联系作者。
评论