Spring Boot 集成 Kafka

Spring Boot 与 Kafka 集成是实现高效消息通报和数据流处理的常见方式。Spring Boot 提供了简化 Kafka 设置和使用的功能，使得集成过程变得更加直观和高效。以下是 Spring Boot 集成 Kafka 的详细步调，包括设置、生产者和斲丧者的实现以及一些高级特性。
1. 添加依赖

首先，你必要在 Spring Boot 项目标 pom.xml 文件中添加 Kafka 相干的依赖。使用 Spring Boot 的起步依赖可以简化设置。

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

复制代码

2. 设置 Kafka

2.1. 设置文件

在 application.properties 或 application.yml 文件中设置 Kafka 相干属性。
application.properties:

1. # Kafka 服务器地址
2. spring.kafka.bootstrap-servers=localhost:9092
4. # Kafka 消费者配置
5. spring.kafka.consumer.group-id=my-group
6. spring.kafka.consumer.auto-offset-reset=earliest
7. spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
8. spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
10. # Kafka 生产者配置
11. spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
12. spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

复制代码

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

复制代码

2.2. Kafka 设置类

在 Spring Boot 中，你可以使用 @Configuration 注解创建一个设置类，来界说 Kafka 的生产者和斲丧者设置。

1. import org.apache.kafka.clients.producer.ProducerConfig;
2. import org.apache.kafka.clients.consumer.ConsumerConfig;
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.kafka.core.ConsumerFactory;
8. import org.springframework.kafka.core.KafkaTemplate;
9. import org.springframework.kafka.core.ProducerFactory;
10. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
11. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
12. import org.springframework.kafka.core.KafkaTemplate;
13. import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
14. import org.springframework.kafka.listener.config.ContainerProperties;
15. import org.springframework.kafka.annotation.EnableKafka;
16. import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
17. import org.springframework.kafka.core.ConsumerFactory;
18. import org.springframework.kafka.core.ProducerFactory;
19. import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
20. import org.springframework.kafka.core.DefaultKafkaProducerFactory;
22. import java.util.HashMap;
23. import java.util.Map;
25. @Configuration
26. @EnableKafka
27. public class KafkaConfig {
29.     @Bean
30.     public ProducerFactory<String, String> producerFactory() {
31.         Map<String, Object> configProps = new HashMap<>();
32.         configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
33.         configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
34.         configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
35.         return new DefaultKafkaProducerFactory<>(configProps);
36.     }
38.     @Bean
39.     public KafkaTemplate<String, String> kafkaTemplate() {
40.         return new KafkaTemplate<>(producerFactory());
41.     }
43.     @Bean
44.     public ConsumerFactory<String, String> consumerFactory() {
45.         Map<String, Object> configProps = new HashMap<>();
46.         configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
47.         configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
48.         configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
49.         configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
50.         return new DefaultKafkaConsumerFactory<>(configProps);
51.     }
53.     @Bean
54.     public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
55.         ConcurrentKafkaListenerContainerFactory<String, String> factory =
56.                 new ConcurrentKafkaListenerContainerFactory<>();
57.         factory.setConsumerFactory(consumerFactory());
58.         return factory;
59.     }
60. }

复制代码

3. 实现 Kafka 生产者

3.1. 生产者服务

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 KafkaProducerService {
8.     @Autowired
9.     private KafkaTemplate<String, String> kafkaTemplate;
11.     private static final String TOPIC = "my_topic";
13.     public void sendMessage(String message) {
14.         kafkaTemplate.send(TOPIC, message);
15.     }
16. }

复制代码

3.2. 控制器示例

1. import org.springframework.beans.factory.annotation.Autowired;
2. import org.springframework.web.bind.annotation.PostMapping;
3. import org.springframework.web.bind.annotation.RequestBody;
4. import org.springframework.web.bind.annotation.RestController;
6. @RestController
7. public class KafkaController {
9.     @Autowired
10.     private KafkaProducerService kafkaProducerService;
12.     @PostMapping("/send")
13.     public void sendMessage(@RequestBody String message) {
14.         kafkaProducerService.sendMessage(message);
15.     }
16. }

复制代码

4. 实现 Kafka 斲丧者

4.1. 斲丧者服务

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

复制代码

5. 高级特性

5.1. 消息事务

Kafka 支持消息事务，确保消息的原子性。
生产者设置：

1. spring.kafka.producer.enable-idempotence=true
2. spring.kafka.producer.transaction-id-prefix=my-transactional-id

复制代码

使用事务：

1. import org.springframework.kafka.core.KafkaTemplate;
2. import org.springframework.kafka.core.ProducerFactory;
3. import org.springframework.kafka.core.TransactionTemplate;
4. import org.springframework.stereotype.Service;
5. import org.springframework.transaction.annotation.Transactional;
7. @Service
8. public class KafkaTransactionalService {
10.     private final KafkaTemplate<String, String> kafkaTemplate;
11.     private final TransactionTemplate transactionTemplate;
13.     public KafkaTransactionalService(KafkaTemplate<String, String> kafkaTemplate, TransactionTemplate transactionTemplate) {
14.         this.kafkaTemplate = kafkaTemplate;
15.         this.transactionTemplate = transactionTemplate;
16.     }
18.     @Transactional
19.     public void sendMessageInTransaction(String message) {
20.         kafkaTemplate.executeInTransaction(t -> {
21.             kafkaTemplate.send("my_topic", message);
22.             return true;
23.         });
24.     }
25. }

复制代码

5.2. 异步发送与回调

异步发送：

1. public void sendMessageAsync(String message) {
2.     kafkaTemplate.send("my_topic", message).addCallback(
3.         result -> System.out.println("Sent message: " + message),
4.         ex -> System.err.println("Failed to send message: " + ex.getMessage())
5.     );
6. }

复制代码

总结

Spring Boot 与 Kafka 的集成使得消息队列的使用变得更加简朴和高效。通过上述步调，你可以轻松地设置 Kafka、实现生产者和斲丧者，并使用 Spring Boot 提供的强大功能来处理消息流。相识 Kafka 的高级特性（如事务和异步处理）能够帮助你更好地满意业务需求，确保系统的高可用性和数据同等性。

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