SpringBoot Kafka发送消息与接收消息实例

媒介 
Kafka的根本工作原理 

 我们将消息的发布（publish）称作 producer(生产者)，将消息的订阅（subscribe）表述为 consumer（消费者），将中间的存储阵列称作 broker(署理)，这样就可以大抵描画出这样一个局面：
生产者将数据生产出来，交给 broker 进行存储，消费者需要消费数据了，就从broker中去拿出数据来，然后完成一系列对数据的处理操纵。 
 1.引入spring-kafka的jar包

在pom.xml里面导入spring-kafka包

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4.     <modelVersion>4.0.0</modelVersion>
5.     <parent>
6.         <groupId>org.springframework.boot</groupId>
7.         <artifactId>spring-boot-starter-parent</artifactId>
8.         <version>2.7.4</version>
9.         <relativePath/> <!-- lookup parent from repository -->
10.     </parent>
11.     <groupId>com.example</groupId>
12.     <artifactId>SpringBootKafka</artifactId>
13.     <version>0.0.1-SNAPSHOT</version>
14.     <name>SpringBootKafka</name>
15.     <description>SpringBootKafka</description>
17.     <properties>
18.         <java.version>1.8</java.version>
19.     </properties>
20.     <dependencies>
21.         <dependency>
22.             <groupId>org.springframework.boot</groupId>
23.             <artifactId>spring-boot-starter-web</artifactId>
24.         </dependency>
26.         <dependency>
27.             <groupId>org.springframework.boot</groupId>
28.             <artifactId>spring-boot-starter-test</artifactId>
29.             <scope>test</scope>
30.         </dependency>
32.         <!-- pom.xml -->
33.         <dependency>
34.             <groupId>org.springframework.kafka</groupId>
35.             <artifactId>spring-kafka</artifactId>
36.         </dependency>
37.         <dependency>
38.             <groupId>log4j</groupId>
39.             <artifactId>log4j</artifactId>
40.             <version>1.2.17</version>
41.         </dependency>
42.         <dependency>
43.             <groupId>org.projectlombok</groupId>
44.             <artifactId>lombok</artifactId>
45.         </dependency>
46.         <dependency>
47.             <groupId>org.apache.kafka</groupId>
48.             <artifactId>kafka-streams</artifactId>
49.         </dependency>
50.     </dependencies>
52.     <repositories>
53.         <repository>
54.             <id>central</id>
55.             <name>aliyun maven</name>
56.             <url>https://maven.aliyun.com/repository/public/</url>
57.             <layout>default</layout>
58.             <!-- 是否开启发布版构件下载 -->
59.             <releases>
60.                 <enabled>true</enabled>
61.             </releases>
62.             <!-- 是否开启快照版构件下载 -->
63.             <snapshots>
64.                 <enabled>false</enabled>
65.             </snapshots>
66.         </repository>
67.     </repositories>
69.     <build>
70.         <plugins>
71.             <plugin>
72.                 <groupId>org.springframework.boot</groupId>
73.                 <artifactId>spring-boot-maven-plugin</artifactId>
74.             </plugin>
75.         </plugins>
76.     </build>
77. </project>

复制代码

2.编写配置文件

在src/main/resources/application.yml里面编写kafka的配置，包括生成者和消费者 

1. spring:
2.   kafka:
3.     bootstrap-servers: 192.168.110.105:9092
4.     #streams:
5.       #application-id: my-streams-app
6.     consumer:
7.       group-id: myGroupId
8.       auto-offset-reset: latest
9.       enable-auto-commit: true
10.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
11.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
12.     producer:
13.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
14.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
15.       retries: 5

复制代码

3.编写生产者

使用org.springframework.kafka.core.KafkaTemplate来发送消息，这里采用了异步方式，获取了消息的处理结果

1. package com.example.springbootkafka.service;
3. import com.example.springbootkafka.entity.User;
4. import com.fasterxml.jackson.core.JsonProcessingException;
5. import com.fasterxml.jackson.databind.ObjectMapper;
6. import lombok.extern.slf4j.Slf4j;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.kafka.core.KafkaTemplate;
9. import org.springframework.kafka.support.SendResult;
10. import org.springframework.stereotype.Service;
11. import org.springframework.util.concurrent.ListenableFuture;
13. import java.util.concurrent.CompletableFuture;
14. import java.util.concurrent.Future;
16. @Slf4j
17. @Service
18. public class KafkaProducer {
20.     private final KafkaTemplate<String, String> kafkaTemplate;
22.     private final ObjectMapper objectMapper;
24.     @Autowired
25.     public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
26.         this.kafkaTemplate = kafkaTemplate;
27.         this.objectMapper = objectMapper;
28.     }
30.     public void sendMessage(String message) {
31.         log.info("KafkaProducer message:{}", message);
32.         //kafkaTemplate.send("test", message).addCallback();
33.         Future<SendResult<String, String>> future = kafkaTemplate.send("test", message);
34.         CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
35.             try {
36.                 future.get(); // 等待原始future完成
37.             } catch (Exception e) {
38.                 throw new RuntimeException(e);
39.             }
40.         });
42. // 使用whenComplete方法
43.         completableFuture.whenComplete((result, ex) -> {
44.             if (ex != null) {
45.                 System.out.println("Error occurred: " + ex.getMessage());
46.                 // 成功发送
47.             } else {
48.                 System.out.println("Completed successfully");
49.             }
50.         });
52.         /*future.whenComplete((result, ex) -> {
53.             if (ex == null) {
54.                 // 成功发送
55.                 RecordMetadata metadata = result.getRecordMetadata();
56.                 System.out.println("Message sent successfully with offset: " + metadata.offset());
57.             } else {
58.                 // 发送失败
59.                 System.err.println("Failed to send message due to: " + ex.getMessage());
60.             }
61.         });*/
64.     }
66.     public void sendUser(User user) throws JsonProcessingException {
67.         //final ProducerRecord<String, String> record = createRecord(data);
69.         //ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", message);
71.         //ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", user);
73.         String userJson = objectMapper.writeValueAsString(user);
74.         ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", userJson);
75.         /*future.addCallback(
76.                 success -> System.out.println("Message sent successfully: " + userJson),
77.                 failure -> System.err.println("Failed to send message: " + failure.getMessage())
78.         );*/
79.         CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
80.             try {
81.                 future.get(); // 等待原始future完成
82.             } catch (Exception e) {
83.                 throw new RuntimeException(e);
84.             }
85.         });
86.         completableFuture.whenComplete((result, ex) -> {
87.             if (ex != null) {
88.                 System.out.println("Error occurred: " + ex.getMessage());
89.                 // 成功发送
90.             } else {
91.                 System.out.println("Completed successfully");
92.             }
93.         });
94.     }
95. }

复制代码

 4.编写消费者

通过org.springframework.kafka.annotation.KafkaListener来监听消息

1. package com.example.springbootkafka.service;
3. import lombok.extern.slf4j.Slf4j;
4. import org.apache.log4j.Logger;
5. import org.springframework.kafka.annotation.KafkaListener;
6. import org.springframework.stereotype.Service;
9. @Slf4j
10. @Service
11. public class KafkaConsumer {
14.     @KafkaListener(topics = "test", groupId = "myGroupId")
15.     public void consume(String message) {
16.         System.out.println("Received message: " + message);
17.         log.info("KafkaConsumer message:{}", message);
18.     }
19. }

复制代码

5.测试消息的生成与发送

1. package com.example.springbootkafka.controller;
3. import com.example.springbootkafka.entity.User;
4. import com.example.springbootkafka.service.KafkaProducer;
5. import com.fasterxml.jackson.core.JsonProcessingException;
6. import lombok.extern.slf4j.Slf4j;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.web.bind.annotation.GetMapping;
9. import org.springframework.web.bind.annotation.RestController;
11. @Slf4j
12. @RestController
13. public class MessageController {
15.     private final KafkaProducer producer;
17.     @Autowired
18.     public MessageController(KafkaProducer producer) {
19.         this.producer = producer;
20.     }
22.     @GetMapping("/send-message")
23.     public String sendMessage() {
24.         log.info("MessageController sendMessage start!");
25.         producer.sendMessage("hello, Kafka!");
26.         log.info("MessageController sendMessage end!");
27.         return "Message sent successfully.";
28.     }
30.     @GetMapping("/send")
31.     public String sendMessage1() {
32.         log.info("MessageController sendMessage1 start!");
33.         User user = User.builder().name("xuxin").dept("IT/DevlopMent").build();
34.         try {
35.             producer.sendUser(user);
36.         } catch (JsonProcessingException e) {
37.             throw new RuntimeException(e);
38.         }
39.         log.info("MessageController sendMessage1 end!");
40.         return "Message sendMessage1 successfully.";
41.     }
42. }

复制代码

 6.检察结果：

 

具体代码见https://gitee.com/dylan_2017/springboot-kafka.git

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