SpringBoot如何使用Kafka来优化接口请求的并发

在Spring Boot中使用 Kafka 来优化接口请求的并发，主要是通过将耗时的任务异步化到Kafka消息队列中来实现。这样，接口可以立即响应客户端，而不需要等待耗时任务完成。

在Spring Boot应用程序中调用Kafka通常涉及使用Spring Kafka库，它提供了与Apache Kafka的高级集成，使得从Spring Boot应用程序中发送和接收消息变得更加简单和直观。

安装Apache Kafka

编写docker-compose.yml

version: '3' services:   zookeeper:     image: wurstmeister/zookeeper     container_name: zookeeper     ports:       - "2181:2181"   kafka:     image: wurstmeister/kafka     container_name: kafka     ports:       - "9092:9092"     environment:       KAFKA_ADVERTISED_HOST_NAME: localhost       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181     volumes:       - /var/run/docker.sock:/var/run/docker.sock 

使用docker compose启动容器

docker-compose up -d

添加依赖

首先，需要在pom.xml中添加Spring Kafka的依赖。

<!-- Spring Kafka --> <dependency>     <groupId>org.springframework.kafka</groupId>     <artifactId>spring-kafka</artifactId> </dependency>

配置Kafka

在application.properties文件中配置Kafka的属性。

# application.properties   spring.kafka.bootstrap-servers=localhost:9092   spring.kafka.consumer.group-id=myGroup   spring.kafka.consumer.auto-offset-reset=earliest   spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer   spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer   spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer   spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

Kafka生产者服务

创建一个服务类来发送消息到Kafka。

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;      public void sendMessage(String topic, String key, String value) {         // 异步发送消息         kafkaTemplate.send(topic, key, value).addCallback(success -> {             System.out.println("Message sent successfully: " + value);         }, failure -> {             System.err.println("Failed to send message: " + value);         });     } }

Kafka消费者服务

创建一个监听器来接收Kafka中的消息。

import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component;  @Component public class KafkaConsumerService {      @KafkaListener(topics = "your-topic-name", groupId = "myGroup")     public void listen(String message) {         // 处理消息（可能是耗时的操作）         System.out.println("Received message in group 'myGroup': " + message);         // 处理耗时操作         ...     } }

控制器

在控制器中调用Kafka生产者服务来发送消息，并立即响应客户端。

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController;    @RestController public class MyController {      @Autowired       private KafkaProducerService kafkaProducerService;      @PostMapping("/send")     public String sendMessage(@RequestParam String message) {         // 发送消息到Kafka，并立即返回响应         kafkaProducerService.sendMessage("your-topic-name", "key1", message);         return "Message sent to Kafka";     } }