Kafka 下载安装及使用总结

1. 下载安装

官网下载地址：Apache Kafka
下载对应的文件

上传到服务器上，解压

1. tar -xzf kafka_2.13-3.7.0.tgz

复制代码

目次结果如下

1. ├── bin
2. │   └── windows
3. ├── config
4. │   └── kraft
5. ├── libs
6. ├── licenses
7. └── site-docs

复制代码

官方文档：Apache Kafka
kafka 有两种启动方式，ZooKeeper 和 KRaft，这里接纳 KRaft 的方式，使用 kraft 目次下的设置文件

1. config
2. ├── connect-console-sink.properties
3. ├── connect-console-source.properties
4. ├── connect-distributed.properties
5. ├── connect-file-sink.properties
6. ├── connect-file-source.properties
7. ├── connect-log4j.properties
8. ├── connect-mirror-maker.properties
9. ├── connect-standalone.properties
10. ├── consumer.properties
11. ├── kraft
12. │   ├── broker.properties
13. │   ├── controller.properties
14. │   └── server.properties
15. ├── log4j.properties
16. ├── producer.properties
17. ├── server.properties
18. ├── tools-log4j.properties
19. ├── trogdor.conf
20. └── zookeeper.properties

复制代码

修改 server.properties 文件
   

• process.roles：KRaft 模式角色
  
  
  
  • broker
    
  • controller
    
  • broker,controller
    
  
  
• node.id：节点 ID，需要为每个节点分配一个唯一的 ID
  
• controller.quorum.voters：Controller 的投票者设置
  
• log.dirs：日记目次
  

  初始化集群，先天生一个 UUID

1. ./bin/kafka-storage.sh random-uuid

复制代码

再实验下令，使用天生的 UUID 完成集群初始化

1. ./bin/kafka-storage.sh format -t thCDFveGRleJro7zTaOOGA -c ./config/kraft/server.properties

复制代码

然后启动 Kafka

1. ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

复制代码

2. Spring Boot 集成

2.1 引入依赖

1. <dependencies>
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter-web</artifactId>
5.         </dependency>
7.         <dependency>
8.             <groupId>org.springframework.kafka</groupId>
9.             <artifactId>spring-kafka</artifactId>
10.             <version>3.1.4</version>
11.             <exclusions>
12.                 <exclusion>
13.                     <groupId>org.apache.kafka</groupId>
14.                     <artifactId>kafka-clients</artifactId>
15.                 </exclusion>
16.             </exclusions>
17.         </dependency>
19.         <dependency>
20.             <groupId>org.apache.kafka</groupId>
21.             <artifactId>kafka-clients</artifactId>
22.             <version>3.7.0</version>
23.         </dependency>
24. </dependencies>

复制代码

2.2 设置文件

1. spring:
2.   kafka:
3.     bootstrap-servers: 127.0.0.1:9092
4.     consumer:
5.       group-id: fable-group

复制代码

2.3 生产者

使用 KafkaTemplate 实现消息的生产

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.kafka.core.KafkaTemplate;
3. import org.springframework.web.bind.annotation.GetMapping;
4. import org.springframework.web.bind.annotation.RestController;
6. @RestController
7. public class ProducerController {
9.     private final KafkaTemplate<String, String> kafkaTemplate;
11.     @Autowired
12.     public ProducerController(KafkaTemplate<String, String> kafkaTemplate) {
13.         this.kafkaTemplate = kafkaTemplate;
14.     }
16.     @GetMapping("/send/message")
17.     public String sendMessage() {
18.         kafkaTemplate.send("test", "Hello, Kafka!");
19.         return "Message sent successfully";
20.     }
21. }

复制代码

2.4 消费者

添加 @KafkaListener 注解，监听消息

1. import org.apache.kafka.clients.consumer.ConsumerRecord;
2. import org.springframework.kafka.annotation.KafkaListener;
3. import org.springframework.stereotype.Service;
5. @Service
6. public class ConsumerService {
8.     @KafkaListener(topics = "test", groupId = "fable-group")
9.     public void listen(ConsumerRecord<String, String> consumerRecord) {
10.         System.out.println("Received message: " + consumerRecord.value());
11.     }
12. }

复制代码

2.5 启动类

添加 @EnableKafka 注解

1. import org.springframework.boot.SpringApplication;
2. import org.springframework.boot.autoconfigure.SpringBootApplication;
3. import org.springframework.kafka.annotation.EnableKafka;
5. @SpringBootApplication
6. @EnableKafka
7. public class FableApplication {
8.     public static void main(String[] args) {
9.         SpringApplication.run(FableApplication.class, args);
10.     }
11. }

复制代码

2.6 测试

访问 /send/message 接口，可以看到控制台打印出接收到的消息

1. Received message: Hello, Kafka!

复制代码

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