SpringBoot使用Kafka来优化接口请求的并发方式

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 up -d

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

# 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

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);
        });
    }
}

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);
        // 处理耗时操作
        ...
    }
}

@Service  
public class KafkaConsumerService {  
  
    @KafkaListener(topics = "your-topic-name", groupId = "myGroup")  
    public void listen(String message) {  
        // 将消息发送到线程池处理  
        executorService.submit(() -> processMessage(message));  
    }  
  
    private void processMessage(String message) {  
        // 处理消息的逻辑  
        System.out.println("Processing message: " + message);  
        // 模拟耗时操作  
        try {  
            Thread.sleep(1000);  
        } catch (InterruptedException e) {  
            Thread.currentThread().interrupt();  
        }  
    }  
  
    // 线程池配置  
    private ExecutorService executorService = Executors.newFixedThreadPool(30);  
  
    // 确保优雅关闭线程池  
    @PreDestroy  
    public void shutdown() {  
        executorService.shutdown();  
        try {  
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {  
                executorService.shutdownNow();  
            }  
        } catch (InterruptedException ex) {  
            executorService.shutdownNow();  
            Thread.currentThread().interrupt();  
        }  
    }  
}

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";
    }
}