Spring Boot 整合 Kafka 发送提醒消息
整合 Spring Boot 和 Kafka 实现发送提醒消息，可以帮助你在分布式系统中实现高效的消息传递和处理。以下是完整的整合过程，包括依赖配置、Kafka 生产者和消费者的实现，以及测试代码。
1. 添加依赖
首先，在 pom.xml 中添加 Kafka 和 Spring for Apache Kafka 的依赖：
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> </dependencies>
2. 配置 Kafka
在 application.yml 中配置 Kafka 相关属性：
spring: kafka: bootstrap-servers: localhost:9092 consumer: group-id: group_id auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer
3. 创建 Kafka 生产者配置类
创建一个 KafkaProducerConfig 类，用于配置 Kafka 生产者：
import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.StringSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.core.ProducerFactory; import java.util.HashMap; import java.util.Map; @Configuration public class KafkaProducerConfig { @Bean public ProducerFactory<String, String> producerFactory() { Map<String, Object> configProps = new HashMap<>(); configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return new DefaultKafkaProducerFactory<>(configProps); } @Bean public KafkaTemplate<String, String> kafkaTemplate() { return new KafkaTemplate<>(producerFactory()); } }
4. 创建 Kafka 消费者配置类
创建一个 KafkaConsumerConfig 类，用于配置 Kafka 消费者：
import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import java.util.HashMap; import java.util.Map; @EnableKafka @Configuration public class KafkaConsumerConfig { @Bean public ConsumerFactory<String, String> consumerFactory() { Map<String, Object> configProps = new HashMap<>(); configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id"); configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); return new DefaultKafkaConsumerFactory<>(configProps); } @Bean public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); return factory; } }
5. 创建 Kafka 生产者服务
创建一个 KafkaProducerService 类，用于发送消息：
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.stereotype.Service; @Service public class KafkaProducerService { @Autowired private KafkaTemplate<String, String> kafkaTemplate; private static final String TOPIC = "reminder_topic"; public void sendMessage(String message) { kafkaTemplate.send(TOPIC, message); } }
6. 创建 Kafka 消费者监听器
创建一个 KafkaConsumerListener 类，用于接收消息：
import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; @Service public class KafkaConsumerListener { @KafkaListener(topics = "reminder_topic", groupId = "group_id") public void listen(String message) { System.out.println("Received Message: " + message); // 在这里处理消息，例如发送提醒 } }
7. 创建控制器发送消息
创建一个 ReminderController 类，通过 REST API 接收请求并发送 Kafka 消息：
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @RestController @RequestMapping("/api/reminder") public class ReminderController { @Autowired private KafkaProducerService producerService; @PostMapping("/send") public String sendReminder(@RequestBody String message) { producerService.sendMessage(message); return "Message sent to Kafka Topic"; } }
8. 测试代码
创建一个简单的测试类 ReminderControllerTest，测试发送提醒消息：
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.web.servlet.MockMvc; @SpringBootTest @AutoConfigureMockMvc public class ReminderControllerTest { @Autowired private MockMvc mockMvc; @Test public void testSendReminder() throws Exception { String message = "This is a test reminder"; mockMvc.perform(post("/api/reminder/send") .content(message) .contentType("text/plain")) .andExpect(status().isOk()); } }
总结
以上是整合 Spring Boot 和 Kafka 发送提醒消息的完整过程。通过上述步骤，你可以实现一个简单的分布式消息系统，发送和接收提醒消息。可以根据需要扩展和优化该系统，例如添加更多的消费者处理逻辑，增加消息处理的可靠性等。
参考资料
通过上述示例代码，你应该能够快速上手并实现 Spring Boot 与 Kafka 的整合。如果有进一步的问题或需要更详细的解释，请随时询问。
