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

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 的启动或停止。

@AutowiredKafkaListenerEndpointRegistry 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";@Beanpublic 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);}@Beanpublic 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);@AutowiredKafkaListenerEndpointRegistry 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 {@AutowiredKafkaListenerAutomation 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启动：