Sprint Boot教程之五十八：动态启动/停止 Kafka 监听器

news2025/1/14 11:53:17

Spring Boot – 动态启动/停止 Kafka 监听器

当 Spring Boot 应用程序启动时，Kafka Listener 的默认行为是开始监听某个主题。但是，有些情况下我们不想在应用程序启动后立即启动它。

要动态启动或停止 Kafka Listener，我们需要三种主要方法，即在需要处理 Kafka 消息时启动/停止、使用@KafkaListener注释、使用 kafkaListenerEndpointRegistry

在本文中，我们将介绍如何动态启动或停止 Kafka 监听器。

启动/停止 Kafka 监听器的不同方法

方法一：

当需要处理 Kafka 消息时，启动一个应用程序。
处理成功后停止应用程序。

方法 2：在注册 Kafka Listener 时，我们可以设置以下 id 属性。

@KafkaListener(id = "id-1", groupId = "group-1", topics = "Message-topic", 
                   containerFactory = "messageListenerFactory", autoStartup = "false")
    public void consumeMessage(Message message)

方法 3：自动连接KafkaListenerEndpointRegistry bean 来控制 Kafka Listener 的启动或停止。

@Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

开始：

 public boolean startListener(String listenerId) {
        MessageListenerContainer listenerContainer = 
          kafkaListenerEndpointRegistry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;
        listenerContainer.start();

停止：

public boolean stopListener(String listenerId) {
        MessageListenerContainer listenerContainer = 
          kafkaListenerEndpointRegistry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;
        listenerContainer.stop();
        logger.info("{} Kafka Listener Stopped.", listenerId);

下面我们将以上述句法方法为例进行实现。

启动或停止特定 Kafka Listener的实现

创建一个类，其对象将被 Kafka 侦听器使用。

文件：Message.java

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@AllArgsConstructor
@NoArgsConstructor
public class Message {
    private String message;
}

配置 Kafka Listener 将使用的消费者。

文件：KakfaConsumerConfig.java

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.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConsumerConfig {

    private String kafkaUrl = "localhost:9092";

    @Bean
    public ConsumerFactory<String, Message> messageConsumerFactory() {
        JsonDeserializer<Message> deserializer = 
             new JsonDeserializer<>(Message.class, false);
        deserializer.setRemoveTypeHeaders(false);
        deserializer.addTrustedPackages("*");
        deserializer.setUseTypeMapperForKey(true);

        Map<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
                   StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
                   JsonDeserializer.class);
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group-1");
        return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(), 
                                                 deserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Message> 
      messageListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Message> containerFactory = 
          new ConcurrentKafkaListenerContainerFactory();
        containerFactory.setConsumerFactory(messageConsumerFactory());
        return containerFactory;
    }
}

创建一个具有必要参数的 Kafka 监听器。

id：此侦听器的容器唯一标识符。如果未指定，则使用自动生成的 ID。
groupId：仅为该监听器使用该值覆盖消费者工厂的 group.id 属性。
主题：此侦听器的主题。条目可以是“主题名称”、“属性占位符键”或“表达式”。主题名称必须从表达式解析。这使用组管理，Kafka 将为组成员分配分区。
containerFactory：KafkaListenerContainerFactory的 bean 名称，将用于创建为该端点提供服务的消息侦听器容器。
autoStartup：设置为 true 或 false 以覆盖容器工厂的默认设置。默认情况下，该值设置为 true，因此，它将在我们的应用程序启动时立即开始使用消息。

文件：KafkaMessageListener.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

@Configuration
public class KafkaMessageListener {

    Logger logger = LoggerFactory.getLogger(KafkaMessageListener.class);

    @KafkaListener(id = "id-1", groupId = "group-1", topics = "Message-topic", 
                   containerFactory = "messageListenerFactory", autoStartup = "false")
    public void consumeMessage(Message message) {
        logger.info("Message received : -> {}", message);
    }
}

KafkaListenerEndpointRegistry 类可用于通过 listenerId 获取 Kafka 侦听器容器。这里我们使用了@KafkaListener注释来将 bean 方法声明为 Kafka 侦听器容器的侦听器。现在可以使用此容器启动或停止 Kafka 侦听器。

文件：KafkaListenerAutomation.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Component;

@Component
public class KafkaListenerAutomation {

    private final Logger logger = LoggerFactory.getLogger(KafkaListenerAutomation.class);

    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    public boolean startListener(String listenerId) {
        MessageListenerContainer listenerContainer = 
           kafkaListenerEndpointRegistry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;
        listenerContainer.start();
        logger.info("{} Kafka Listener Started", listenerId);
        return true;
    }

    public boolean stopListener(String listenerId) {
        MessageListenerContainer listenerContainer = 
          kafkaListenerEndpointRegistry.getListenerContainer(listenerId);
        assert listenerContainer != null : false;
        listenerContainer.stop();
        logger.info("{} Kafka Listener Stopped.", listenerId);
        return true;
    }
}

使用 API 端点，我们可以通过提供 listenerID 来启动或停止特定的 Kafka 监听器。

文件：StartOrStopListenerController.java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class StartOrStopListenerController {

    @Autowired
    KafkaListenerAutomation kafkaListenerAutomation;

    @GetMapping("/start")
    public void start(@RequestParam("id") String listenerId) {
        kafkaListenerAutomation.startListener(listenerId);
    }

    @GetMapping("/stop")
    public void stop(@RequestParam("id") String listenerId) {
        kafkaListenerAutomation.stopListener(listenerId);
    }
}

输出：

1.Kafka Listener启动：