Spring Boot 项目中集成 Kafka-03

在 Spring Boot 项目中集成 Kafka 有多种方式，顺应不同的应用场景和需求。以下将详细先容几种常用的集成方法，包罗：

• 使用 Spring Kafka (KafkaTemplate 和 @KafkaListener)
  
• 使用 Spring Cloud Stream 与 Kafka Binder
  
• 使用 Spring for Apache Kafka Reactive（基于 Reactor）
  
• 手动配置 Producer 和 Consumer Bean
  
• 使用 Spring Integration Kafka
  
• 在测试中使用嵌入式 Kafka
  

每种方法都有其特点和适用场景，选择合适的方法可以或许有效提拔开发效率和应用性能。

---

1. 使用 Spring Kafka (KafkaTemplate 和 @KafkaListener)

这是最常用的 Spring Boot 集成 Kafka 的方式，依赖于 Spring for Apache Kafka 项目，提供了 KafkaTemplate 用于发送消息，以及 @KafkaListener 注解用于接收消息。
步骤一：添加 Maven 依赖

在 pom.xml 中引入 spring-kafka 依赖：

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4. </dependency>

复制代码

步骤二：配置 application.properties 或 application.yml

示例 (application.properties)：

1. # Kafka 集群地址
2. spring.kafka.bootstrap-servers=worker1:9092,worker2:9092,worker3:9092
4. # 生产者配置
5. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
6. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
7. spring.kafka.producer.acks=1
8. spring.kafka.producer.retries=3
9. spring.kafka.producer.batch-size=16384
10. spring.kafka.producer.linger.ms=1
12. # 消费者配置
13. spring.kafka.consumer.group-id=myGroup
14. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
15. spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
16. spring.kafka.consumer.auto-offset-reset=earliest
17. spring.kafka.consumer.enable-auto-commit=true

复制代码

步骤三：编写消息生产者

使用 KafkaTemplate 发送消息：

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.kafka.core.KafkaTemplate;
3. import org.springframework.stereotype.Service;
5. @Service
6. public class ProducerService {
8.     @Autowired
9.     private KafkaTemplate<String, String> kafkaTemplate;
11.     private static final String TOPIC = "topic1";
13.     public void sendMessage(String message) {
14.         kafkaTemplate.send(TOPIC, message);
15.     }
16. }

复制代码

步骤四：编写消息消费者

使用 @KafkaListener 接收消息：

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

复制代码

优缺点

• 长处：
  
  
  
  • 简单易用，快速上手。
    
  • 与 Spring 生态体系无缝集成。
    
  • 支持事务、幂等性等高级特性。
    
  
  
• 缺点：
  
  
  
  • 适用于传统的阻塞式应用，若需要响应式编程则不够友好。
    
  
  

---

2. 使用 Spring Cloud Stream 与 Kafka Binder

Spring Cloud Stream 是一个构建消息驱动微服务的框架，通过 Binder（绑定器）与不同的消息中间件集成。使用 Kafka Binder，可以更加简化 Kafka 与 Spring Boot 的集成。
步骤一：添加 Maven 依赖

在 pom.xml 中引入 spring-cloud-starter-stream-kafka 依赖：

1. <dependency>
2.     <groupId>org.springframework.cloud</groupId>
3.     <artifactId>spring-cloud-starter-stream-kafka</artifactId>
4. </dependency>

复制代码

并确保引入 Spring Cloud 的 BOM 以管理版本：

1. <dependencyManagement>
2.     <dependencies>
3.         <dependency>
4.             <groupId>org.springframework.cloud</groupId>
5.             <artifactId>spring-cloud-dependencies</artifactId>
6.             <version>Hoxton.SR12</version>
7.             <type>pom</type>
8.             <scope>import</scope>
9.         </dependency>
10.     </dependencies>
11. </dependencyManagement>

复制代码

步骤二：配置 application.yml

1. spring:
2.   cloud:
3.     stream:
4.       bindings:
5.         output:
6.           destination: topic1
7.           contentType: application/json
8.         input:
9.           destination: topic1
10.           group: myGroup
11.       kafka:
12.         binder:
13.           brokers: worker1:9092,worker2:9092,worker3:9092

复制代码

步骤三：编写消息生产者

使用 @EnableBinding 和 Source 接口：

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.cloud.stream.annotation.EnableBinding;
3. import org.springframework.cloud.stream.messaging.Source;
4. import org.springframework.messaging.support.MessageBuilder;
5. import org.springframework.stereotype.Service;
7. @Service
8. @EnableBinding(Source.class)
9. public class StreamProducerService {
11.     @Autowired
12.     private Source source;
14.     public void sendMessage(String message) {
15.         source.output().send(MessageBuilder.withPayload(message).build());
16.     }
17. }

复制代码

步骤四：编写消息消费者

使用 @StreamListener 注解：

1. import org.springframework.cloud.stream.annotation.EnableBinding;
2. import org.springframework.cloud.stream.annotation.StreamListener;
3. import org.springframework.cloud.stream.messaging.Sink;
4. import org.springframework.stereotype.Component;
6. @Component
7. @EnableBinding(Sink.class)
8. public class StreamConsumerService {
10.     @StreamListener(Sink.INPUT)
11.     public void handleMessage(String message) {
12.         System.out.println("Received message: " + message);
13.     }
14. }

复制代码

优缺点

• 长处：
  
  
  
  • 高度抽象，淘汰配置与代码量。
    
  • 更得当微服务架构，支持绑定多个输入输出。
    
  • 支持多种消息中间件，易于切换。
    
  
  
• 缺点：
  
  
  
  • 抽象层较高，大概难以实现一些细粒度的自定义配置。
    
  • 学习曲线较陡，需明白 Binder 和 Channel 的概念。
    
  
  

---

3. 使用 Spring for Apache Kafka Reactive（基于 Reactor）

对于需要响应式编程的应用，可以使用基于 Reactor 的 Spring Kafka Reactive 举行集成，实现非阻塞的消息处理。
步骤一：添加 Maven 依赖

目前，Spring Kafka 本身并未直接提供响应式支持，但可以结合 Project Reactor Kafka 使用。
引入 Reactor Kafka 依赖：

1. <dependency>
2.     <groupId>io.projectreactor.kafka</groupId>
3.     <artifactId>reactor-kafka</artifactId>
4.     <version>1.3.11</version>
5. </dependency>

复制代码

步骤二：配置 application.yml

1. kafka:
2.   bootstrap-servers: worker1:9092,worker2:9092,worker3:9092
3.   producer:
4.     key-serializer: org.apache.kafka.common.serialization.StringSerializer
5.     value-serializer: org.apache.kafka.common.serialization.StringSerializer
6.   consumer:
7.     group-id: myReactiveGroup
8.     key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
9.     value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
10.     auto-offset-reset: earliest

复制代码

步骤三：编写响应式消息生产者

使用 SenderOptions 和 KafkaSender：

1. import org.apache.kafka.clients.producer.ProducerConfig;
2. import org.apache.kafka.common.serialization.StringSerializer;
3. import org.springframework.beans.factory.annotation.Value;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.stereotype.Service;
6. import reactor.kafka.sender.KafkaSender;
7. import reactor.kafka.sender.SenderOptions;
8. import reactor.kafka.sender.SenderRecord;
9. import reactor.core.publisher.Mono;
11. import java.util.HashMap;
12. import java.util.Map;
14. @Service
15. public class ReactiveProducerService {
17.     @Value("${kafka.bootstrap-servers}")
18.     private String bootstrapServers;
20.     @Bean
21.     public SenderOptions<String, String> senderOptions() {
22.         Map<String, Object> props = new HashMap<>();
23.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
24.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
25.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
26.         return SenderOptions.create(props);
27.     }
29.     @Bean
30.     public KafkaSender<String, String> kafkaSender(SenderOptions<String, String> senderOptions) {
31.         return KafkaSender.create(senderOptions);
32.     }
34.     public Mono<Void> sendMessage(String topic, String key, String value) {
35.         SenderRecord<String, String, Integer> record = SenderRecord.create(
36.                 new org.apache.kafka.clients.producer.ProducerRecord<>(topic, key, value),
37.                 1
38.         );
39.         return kafkaSender(senderOptions())
40.                 .send(Mono.just(record))
41.                 .then();
42.     }
43. }

复制代码

步骤四：编写响应式消息消费者

使用 ReceiverOptions 和 KafkaReceiver：

1. import org.apache.kafka.clients.consumer.ConsumerConfig;
2. import org.apache.kafka.common.serialization.StringDeserializer;
3. import org.springframework.beans.factory.annotation.Value;
4. import org.springframework.context.annotation.Bean;
5. import org.springframework.stereotype.Service;
6. import reactor.core.publisher.Flux;
7. import reactor.kafka.receiver.KafkaReceiver;
8. import reactor.kafka.receiver.ReceiverOptions;
9. import reactor.kafka.receiver.ReceiverRecord;
11. import java.util.HashMap;
12. import java.util.Map;
14. @Service
15. public class ReactiveConsumerService {
17.     @Value("${kafka.bootstrap-servers}")
18.     private String bootstrapServers;
20.     @Value("${kafka.consumer.group-id}")
21.     private String groupId;
23.     @Bean
24.     public ReceiverOptions<String, String> receiverOptions() {
25.         Map<String, Object> props = new HashMap<>();
26.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
27.         props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
28.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
29.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
30.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
31.         ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(props);
32.         return receiverOptions.subscription(java.util.Collections.singleton("topic1"));
33.     }
35.     @Bean
36.     public Flux<ReceiverRecord<String, String>> kafkaFlux(ReceiverOptions<String, String> receiverOptions) {
37.         return KafkaReceiver.create(receiverOptions).receive();
38.     }
40.     public void consumeMessages() {
41.         kafkaFlux(receiverOptions())
42.                 .doOnNext(record -> {
43.                     System.out.println("Received: " + record.value());
44.                     record.receiverOffset().acknowledge();
45.                 })
46.                 .subscribe();
47.     }
48. }

复制代码

优缺点

• 长处：
  
  
  
  • 支持响应式编程模子，适用于高并发和非阻塞场景。
    
  • 更高的资源使用率和吞吐量。
    
  
  
• 缺点：
  
  
  
  • 相较于传统阻塞式，开发复杂度更高。
    
  • 需要明白 Reactor 和响应式编程的基本概念。
    
  
  

---

4. 手动配置 Producer 和 Consumer Bean

对于需要更高自定义配置的应用，可以手动配置 ProducerFactory, ConsumerFactory, KafkaTemplate 和 ConcurrentKafkaListenerContainerFactory 等 Bean。
步骤一：添加 Maven 依赖

在 pom.xml 中引入 spring-kafka 依赖：

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4. </dependency>

复制代码

步骤二：编写 Kafka 配置类

1. import org.apache.kafka.clients.consumer.ConsumerConfig;
2. import org.apache.kafka.clients.producer.ProducerConfig;
3. import org.apache.kafka.common.serialization.StringDeserializer;
4. import org.apache.kafka.common.serialization.StringSerializer;
5. import org.springframework.beans.factory.annotation.Value;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
8. import org.springframework.kafka.annotation.EnableKafka;
9. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
10. import org.springframework.kafka.core.*;
12. import java.util.HashMap;
13. import java.util.Map;
15. @Configuration
16. @EnableKafka
17. public class KafkaManualConfig {
19.     @Value("${kafka.bootstrap-servers}")
20.     private String bootstrapServers;
22.     @Value("${kafka.consumer.group-id}")
23.     private String groupId;
25.     // Producer 配置
26.     @Bean
27.     public ProducerFactory<String, String> producerFactory() {
28.         Map<String, Object> configProps = new HashMap<>();
29.         configProps.put(
30.                 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
31.                 bootstrapServers);
32.         configProps.put(
33.                 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
34.                 StringSerializer.class);
35.         configProps.put(
36.                 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
37.                 StringSerializer.class);
38.         // 其他自定义配置
39.         return new DefaultKafkaProducerFactory<>(configProps);
40.     }
42.     @Bean
43.     public KafkaTemplate<String, String> kafkaTemplate() {
44.         return new KafkaTemplate<>(producerFactory());
45.     }
47.     // Consumer 配置
48.     @Bean
49.     public ConsumerFactory<String, String> consumerFactory() {
50.         Map<String, Object> props = new HashMap<>();
51.         props.put(
52.                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
53.                 bootstrapServers);
54.         props.put(
55.                 ConsumerConfig.GROUP_ID_CONFIG,
56.                 groupId);
57.         props.put(
58.                 ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
59.                 StringDeserializer.class);
60.         props.put(
61.                 ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
62.                 StringDeserializer.class);
63.         // 其他自定义配置
64.         return new DefaultKafkaConsumerFactory<>(props);
65.     }
67.     // KafkaListenerContainerFactory
68.     @Bean
69.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
70.         ConcurrentKafkaListenerContainerFactory<String, String> factory =
71.                 new ConcurrentKafkaListenerContainerFactory<>();
72.         factory.setConsumerFactory(consumerFactory());
73.         // 其他自定义配置，如并发数、批量消费等
74.         return factory;
75.     }
76. }

复制代码

步骤三：编写消息生产者和消费者

Producer 示例：

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.kafka.core.KafkaTemplate;
3. import org.springframework.stereotype.Service;
5. @Service
6. public class ManualProducerService {
8.     @Autowired
9.     private KafkaTemplate<String, String> kafkaTemplate;
11.     private static final String TOPIC = "topic1";
13.     public void sendMessage(String message) {
14.         kafkaTemplate.send(TOPIC, message);
15.     }
16. }

复制代码

Consumer 示例：

1. import org.springframework.kafka.annotation.KafkaListener;
2. import org.springframework.stereotype.Component;
4. @Component
5. public class ManualConsumerService {
7.     @KafkaListener(topics = "topic1", groupId = "myGroup")
8.     public void listen(String message) {
9.         System.out.println("Received message: " + message);
10.     }
11. }

复制代码

优缺点

• 长处：
  
  
  
  • 高度自定义，适用于复杂配置需求。
    
  • 可以机动配置多个 KafkaTemplate 或 KafkaListenerContainerFactory，顺应多种场景。
    
  
  
• 缺点：
  
  
  
  • 配置较为繁琐，代码量增加。
    
  • 需要深入明白 Spring Kafka 的配置与使用。
    
  
  

---

5. 使用 Spring Integration Kafka

Spring Integration 提供了对 Kafka 的集成支持，适用于需要集成多种消息渠道和复杂消息路由的应用。
步骤一：添加 Maven 依赖

在 pom.xml 中引入 spring-integration-kafka 依赖：

1. <dependency>
2.     <groupId>org.springframework.integration</groupId>
3.     <artifactId>spring-integration-kafka</artifactId>
4.     <version>3.3.5.RELEASE</version>
5. </dependency>

复制代码

步骤二：编写 Kafka Integration 配置

1. import org.apache.kafka.clients.consumer.ConsumerConfig;
2. import org.apache.kafka.clients.producer.ProducerConfig;
3. import org.apache.kafka.common.serialization.StringDeserializer;
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.integration.annotation.ServiceActivator;
8. import org.springframework.integration.channel.DirectChannel;
9. import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
10. import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
11. import org.springframework.kafka.core.*;
12. import org.springframework.messaging.MessageChannel;
13. import org.springframework.messaging.MessageHandler;
15. import java.util.HashMap;
16. import java.util.Map;
18. @Configuration
19. public class SpringIntegrationKafkaConfig {
21.     @Bean
22.     public ProducerFactory<String, String> producerFactory() {
23.         Map<String, Object> props = new HashMap<>();
24.         props.put(
25.                 ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
26.                 "worker1:9092,worker2:9092,worker3:9092");
27.         props.put(
28.                 ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
29.                 StringSerializer.class);
30.         props.put(
31.                 ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
32.                 StringSerializer.class);
33.         return new DefaultKafkaProducerFactory<>(props);
34.     }
36.     @Bean
37.     public KafkaTemplate<String, String> kafkaTemplate() {
38.         return new KafkaTemplate<>(producerFactory());
39.     }
41.     // 消费者工厂
42.     @Bean
43.     public ConsumerFactory<String, String> consumerFactory() {
44.         Map<String, Object> props = new HashMap<>();
45.         props.put(
46.                 ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
47.                 "worker1:9092,worker2:9092,worker3:9092");
48.         props.put(
49.                 ConsumerConfig.GROUP_ID_CONFIG,
50.                 "myGroup");
51.         props.put(
52.                 ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
53.                 StringDeserializer.class);
54.         props.put(
55.                 ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
56.                 StringDeserializer.class);
57.         return new DefaultKafkaConsumerFactory<>(props);
58.     }
60.     // 输入通道
61.     @Bean
62.     public MessageChannel inputChannel() {
63.         return new DirectChannel();
64.     }
66.     // 消费者适配器
67.     @Bean
68.     public KafkaMessageDrivenChannelAdapter<String, String> kafkaMessageDrivenChannelAdapter() {
69.         KafkaMessageDrivenChannelAdapter<String, String> adapter =
70.                 new KafkaMessageDrivenChannelAdapter<>(consumerFactory(), "topic1");
71.         adapter.setOutputChannel(inputChannel());
72.         return adapter;
73.     }
75.     // 消费者处理器
76.     @Bean
77.     @ServiceActivator(inputChannel = "inputChannel")
78.     public MessageHandler messageHandler() {
79.         return message -> {
80.             String payload = (String) message.getPayload();
81.             System.out.println("Received message: " + payload);
82.         };
83.     }
85.     // 输出通道
86.     @Bean
87.     public MessageChannel outputChannel() {
88.         return new DirectChannel();
89.     }
91.     // 生产者处理器
92.     @Bean
93.     @ServiceActivator(inputChannel = "outputChannel")
94.     public MessageHandler producerMessageHandler(KafkaTemplate<String, String> kafkaTemplate) {
95.         KafkaProducerMessageHandler<String, String> handler =
96.                 new KafkaProducerMessageHandler<>(kafkaTemplate);
97.         handler.setTopicExpressionString("'topic1'");
98.         return handler;
99.     }
100. }

复制代码

步骤三：发送消息到输出通道

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.messaging.MessageChannel;
3. import org.springframework.messaging.support.MessageBuilder;
4. import org.springframework.stereotype.Service;
6. @Service
7. public class IntegrationProducerService {
9.     @Autowired
10.     private MessageChannel outputChannel;
12.     public void sendMessage(String message) {
13.         outputChannel.send(MessageBuilder.withPayload(message).build());
14.     }
15. }

复制代码

优缺点

• 长处：
  
  
  
  • 强盛的消息路由和转换功能，适用于复杂集成场景。
    
  • 可以与 Spring Integration 的其他模块无缝协作。
    
  
  
• 缺点：
  
  
  
  • 配置复杂，学习成本较高。
    
  • 对于简单的 Kafka 集成场景，大概显得过于痴肥。
    
  
  

---

6. 在测试中使用嵌入式 Kafka

在集成测试中，使用嵌入式 Kafka 可以避免依赖外部 Kafka 集群，提拔测试效率与稳固性。
步骤一：添加 Maven 依赖

在 pom.xml 中引入 spring-kafka-test 依赖：

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka-test</artifactId>
4.     <scope>test</scope>
5. </dependency>

复制代码

步骤二：编写测试类

使用 @EmbeddedKafka 注解启动嵌入式 Kafka：

1. import org.apache.kafka.clients.consumer.Consumer;
2. import org.apache.kafka.clients.consumer.ConsumerConfig;
3. import org.apache.kafka.common.serialization.StringDeserializer;
4. import org.junit.jupiter.api.AfterAll;
5. import org.junit.jupiter.api.BeforeAll;
6. import org.junit.jupiter.api.Test;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.boot.test.context.SpringBootTest;
9. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
10. import org.springframework.kafka.test.EmbeddedKafkaBroker;
11. import org.springframework.kafka.test.context.EmbeddedKafka;
12. import org.springframework.kafka.test.utils.KafkaTestUtils;
14. import java.util.Map;
16. @SpringBootTest
17. @EmbeddedKafka(partitions = 1, topics = { "topic1" }, brokerProperties = { "listeners=PLAINTEXT://localhost:9092", "port=9092" })
18. public class KafkaIntegrationTest {
20.     @Autowired
21.     private EmbeddedKafkaBroker embeddedKafkaBroker;
23.     private static Consumer<String, String> consumer;
25.     @BeforeAll
26.     public static void setUp(@Autowired EmbeddedKafkaBroker embeddedKafkaBroker) {
27.         Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGroup", "true", embeddedKafkaBroker);
28.         consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
29.         consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
30.         DefaultKafkaConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
31.         consumer = consumerFactory.createConsumer();
32.         embeddedKafkaBroker.consumeFromAnEmbeddedTopic(consumer, "topic1");
33.     }
35.     @AfterAll
36.     public static void tearDown() {
37.         if (consumer != null) {
38.             consumer.close();
39.         }
40.     }
42.     @Test
43.     public void testSendAndReceive() {
44.         // 发送消息
45.         // 假设有一个 ProducerService 可以发送消息
46.         // producerService.sendMessage("Hello, Kafka!");
48.         // 接收消息
49.         // Consumer Record 验证逻辑
50.         // 可以使用 KafkaTestUtils 来接收消息并断言
51.     }
52. }

复制代码

优缺点

• 长处：
  
  
  
  • 不依赖外部 Kafka 集群，得当 CI/CD 情况。
    
  • 提拔测试的可重复性与稳固性。
    
  
  
• 缺点：
  
  
  
  • 嵌入式 Kafka 启动较慢，大概影响测试速率。
    
  • 需要额外配置，测试代码复杂度增加。
    
  
  

---

总结

在 Spring Boot 中集成 Kafka 有多种方式，每种方式适用于不同的应用场景和需求：

• Spring Kafka (KafkaTemplate 和 @KafkaListener)
  适用于大多数常规应用，简单易用，与 Spring 生态体系无缝集成。
  
• Spring Cloud Stream 与 Kafka Binder
  适用于微服务架构，需处理复杂消息路由与多中间件支持的场景。
  
• Spring for Apache Kafka Reactive
  适用于需要响应式编程模子、高并发和非阻塞消息处理的应用。
  
• 手动配置 Producer 和 Consumer Bean
  适用于需要高度自定义 Kafka 配置和行为的应用。
  
• Spring Integration Kafka
  适用于复杂集成场景，需要与其他消息渠道和体系协作的应用。
  
• 嵌入式 Kafka 在测试中使用
  适用于编写集成测试，提拔测试效率和稳固性。
  

根据项目的具体需求，选择最合适的集成方式可以或许有效提拔开发效率，确保应用的稳固性与可扩展性。

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