Spring Boot 整合 Kafka 详解

Spring Boot 整合 Kafka 详解

本文将详细介绍怎样在 Spring Boot 项目中整合 Apache Kafka，包括 Kafka 的设置、消息的同步和异步发送。
1. 情况准备

在开始之前，请确保你已经安装并设置好 Kafka 集群。假如还没有，请参考 Kafka 官方文档进行安装和设置。
2. 创建 Spring Boot 项目

2.1 使用 Spring Initializr 创建项目

访问 Spring Initializr，选择以下设置：

• Project: Maven Project
  
• Language: Java
  
• Spring Boot: 2.2.2.RELEASE
  
• Dependencies: Spring for Apache Kafka
  

点击 “Generate” 按钮，下载生成的项目，并解压到本地。
3. 添加依靠

在 pom.xml 文件中添加 Kafka 依靠：

1. <dependencies>
2.     <dependency>
3.         <groupId>org.springframework.boot</groupId>
4.         <artifactId>spring-boot-starter</artifactId>
5.     </dependency>
6.     <dependency>
7.         <groupId>org.springframework.kafka</groupId>
8.         <artifactId>spring-kafka</artifactId>
9.     </dependency>
10. </dependencies>

复制代码

4. 设置 Kafka

在 src/main/resources 目次下创建 application.yml 文件，并添加以下设置：

1. spring:
2.   kafka:
3.     bootstrap-servers: localhost:9092
4.     consumer:
5.       group-id: my-group
6.       auto-offset-reset: earliest
7.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
8.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
9.     producer:
10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer

复制代码

设置阐明：

• bootstrap-servers: Kafka broker 的地址列表。
  
• consumer: 斲丧者设置，包括斲丧者组 ID、偏移量重置策略、键和值的反序列化器。
  
• producer: 生产者设置，包括键和值的序列化器。
  

5. 创建 Kafka 生产者

在 src/main/java/com/example/demo 目次下创建 KafkaProducerConfig.java 文件，并添加以下代码：

1. package com.example.demo;
3. import org.apache.kafka.clients.producer.ProducerConfig;
4. import org.apache.kafka.common.serialization.StringSerializer;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
7. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
8. import org.springframework.kafka.core.KafkaTemplate;
9. import org.springframework.kafka.core.ProducerFactory;
10. import org.springframework.kafka.core.ProducerFactory;
11. import org.springframework.kafka.core.ProducerFactory;
12. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
14. import java.util.HashMap;
15. import java.util.Map;
17. @Configuration
18. public class KafkaProducerConfig {
20.     @Bean
21.     public ProducerFactory<String, String> producerFactory() {
22.         Map<String, Object> configProps = new HashMap<>();
23.         configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
24.         configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
25.         configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
26.         return new DefaultKafkaProducerFactory<>(configProps);
27.     }
29.     @Bean
30.     public KafkaTemplate<String, String> kafkaTemplate() {
31.         return new KafkaTemplate<>(producerFactory());
32.     }
33. }

复制代码

6. 发送消息

在 src/main/java/com/example/demo 目次下创建 KafkaProducerService.java 文件，并添加以下代码：

1. package com.example.demo;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.kafka.core.KafkaTemplate;
5. import org.springframework.kafka.support.SendResult;
6. import org.springframework.stereotype.Service;
7. import org.springframework.util.concurrent.ListenableFuture;
8. import org.springframework.util.concurrent.ListenableFutureCallback;
10. @Service
11. public class KafkaProducerService {
13.     @Autowired
14.     private KafkaTemplate<String, String> kafkaTemplate;
16.     private static final String TOPIC = "my-topic";
18.     // 同步发送消息
19.     public void sendMessageSync(String message) {
20.         try {
21.             kafkaTemplate.send(TOPIC, message).get();
22.             System.out.println("同步消息发送成功: " + message);
23.         } catch (Exception e) {
24.             e.printStackTrace();
25.             System.out.println("同步消息发送失败: " + message);
26.         }
27.     }
29.     // 异步发送消息
30.     public void sendMessageAsync(String message) {
31.         ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, message);
32.         future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
33.             @Override
34.             public void onSuccess(SendResult<String, String> result) {
35.                 System.out.println("异步消息发送成功: " + message);
36.             }
38.             @Override
39.             public void onFailure(Throwable ex) {
40.                 ex.printStackTrace();
41.                 System.out.println("异步消息发送失败: " + message);
42.             }
43.         });
44.     }
45. }

复制代码

7. 测试 Kafka 生产者

在 src/main/java/com/example/demo 目次下创建 DemoApplication.java 文件，并添加以下代码：

1. package com.example.demo;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.boot.CommandLineRunner;
5. import org.springframework.boot.SpringApplication;
6. import org.springframework.boot.autoconfigure.SpringBootApplication;
8. @SpringBootApplication
9. public class DemoApplication implements CommandLineRunner {
11.     @Autowired
12.     private KafkaProducerService kafkaProducerService;
14.     public static void main(String[] args) {
15.         SpringApplication.run(DemoApplication.class, args);
16.     }
18.     @Override
19.     public void run(String... args) throws Exception {
20.         kafkaProducerService.sendMessageSync("Hello, Kafka (Sync)!");
21.         kafkaProducerService.sendMessageAsync("Hello, Kafka (Async)!");
22.     }
23. }

复制代码

8. 运行结果

运行 DemoApplication 类，将看到控制台输出如下消息：

1. 同步消息发送成功: Hello, Kafka (Sync)!
2. 异步消息发送成功: Hello, Kafka (Async)!

复制代码

假如 Kafka 生产者发送消息失败，将看到错误信息。
9. 总结

本文详细介绍了怎样在 Spring Boot 项目中整合 Apache Kafka，包括 Kafka 的设置、消息的同步和异步发送。通过明确和实践这些内容，可以资助你更好地掌握 Spring Boot 与 Kafka 的整合与应用。渴望本文对你有所资助，如有任何疑问或建议，欢迎留言讨论。

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