快捷導(dǎo)航

SpringBoot使用Kafka來(lái)優(yōu)化接口請(qǐng)求的并發(fā)方式

更新時(shí)間：2024年07月30日 14:58:57 作者：培根芝士

這篇文章主要介紹了SpringBoot使用Kafka來(lái)優(yōu)化接口請(qǐng)求的并發(fā)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

在Spring Boot中使用 Kafka 來(lái)優(yōu)化接口請(qǐng)求的并發(fā)，主要是通過(guò)將耗時(shí)的任務(wù)異步化到Kafka消息隊(duì)列中來(lái)實(shí)現(xiàn)。這樣，接口可以立即響應(yīng)客戶端，而不需要等待耗時(shí)任務(wù)完成。

在Spring Boot應(yīng)用程序中調(diào)用Kafka通常涉及使用Spring Kafka庫(kù)，它提供了與Apache Kafka的高級(jí)集成，使得從Spring Boot應(yīng)用程序中發(fā)送和接收消息變得更加簡(jiǎn)單和直觀。

安裝Apache Kafka

編寫(xiě)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啟動(dòng)容器

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生產(chǎn)者服務(wù)

創(chuàng)建一個(gè)服務(wù)類來(lái)發(fā)送消息到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) {
        // 異步發(fā)送消息
        kafkaTemplate.send(topic, key, value).addCallback(success -> {
            System.out.println("Message sent successfully: " + value);
        }, failure -> {
            System.err.println("Failed to send message: " + value);
        });
    }
}

kafkaTemplate.send 方法是 Spring Kafka 提供的一個(gè)非常靈活的方法，它允許以不同的方式發(fā)送消息到 Kafka 集群。

當(dāng)調(diào)用 kafkaTemplate.send 方法時(shí)，可以指定要發(fā)送到的 topic、key 和 value，但 key 是可選的。

未指定 key：當(dāng)不指定 key 時(shí)，Kafka 會(huì)根據(jù)配置的分區(qū)器（默認(rèn)是 DefaultPartitioner）來(lái)決定消息應(yīng)該被發(fā)送到哪個(gè)分區(qū)。在沒(méi)有 key 的情況下，分區(qū)器可能會(huì)采用輪詢（round-robin）或其他算法來(lái)隨機(jī)選擇一個(gè)分區(qū)進(jìn)行消息發(fā)送。這種方式下，消息的分布可能會(huì)比較均勻，但無(wú)法控制具有相同邏輯標(biāo)識(shí)的消息被發(fā)送到同一個(gè)分區(qū)。
指定 key：當(dāng)指定 key 時(shí)，Kafka 會(huì)根據(jù) key 的哈希值來(lái)計(jì)算分區(qū)號(hào)，確保具有相同 key 的消息被發(fā)送到同一個(gè)分區(qū)。這種方式有助于保持消息的順序性，因?yàn)?Kafka 保證同一個(gè)分區(qū)內(nèi)的消息是有序的。

Kafka消費(fèi)者服務(wù)

創(chuàng)建一個(gè)監(jiān)聽(tīng)器來(lái)接收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) {
        // 處理消息（可能是耗時(shí)的操作）
        System.out.println("Received message in group 'myGroup': " + message);
        // 處理耗時(shí)操作
        ...
    }
}

在使用Kafka消費(fèi)者時(shí)，Kafka本身已經(jīng)設(shè)計(jì)為支持并發(fā)消費(fèi)，即可以通過(guò)配置多個(gè)消費(fèi)者實(shí)例（partitions的數(shù)量通常決定了并行度的一個(gè)上限，因?yàn)镵afka會(huì)盡量將不同的partitions分配給不同的消費(fèi)者以提高并行度）來(lái)實(shí)現(xiàn)并行處理。

但是，如果想要在消費(fèi)者內(nèi)部進(jìn)一步提高處理消息的并發(fā)度，可以結(jié)合使用Kafka消費(fèi)者和Java的線程池來(lái)實(shí)現(xiàn)。

@Service  
public class KafkaConsumerService {  
  
    @KafkaListener(topics = "your-topic-name", groupId = "myGroup")  
    public void listen(String message) {  
        // 將消息發(fā)送到線程池處理  
        executorService.submit(() -> processMessage(message));  
    }  
  
    private void processMessage(String message) {  
        // 處理消息的邏輯  
        System.out.println("Processing message: " + message);  
        // 模擬耗時(shí)操作  
        try {  
            Thread.sleep(1000);  
        } catch (InterruptedException e) {  
            Thread.currentThread().interrupt();  
        }  
    }  
  
    // 線程池配置  
    private ExecutorService executorService = Executors.newFixedThreadPool(30);  
  
    // 確保優(yōu)雅關(guān)閉線程池  
    @PreDestroy  
    public void shutdown() {  
        executorService.shutdown();  
        try {  
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {  
                executorService.shutdownNow();  
            }  
        } catch (InterruptedException ex) {  
            executorService.shutdownNow();  
            Thread.currentThread().interrupt();  
        }  
    }  
}

控制器

在控制器中調(diào)用Kafka生產(chǎn)者服務(wù)來(lái)發(fā)送消息，并立即響應(yīng)客戶端。

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) {
        // 發(fā)送消息到Kafka，并立即返回響應(yīng)
        kafkaProducerService.sendMessage("your-topic-name", "key1", message);
        return "Message sent to Kafka";
    }
}