在 Spring Boot 中动态管理 Kafka Listener-工具盒子

1、概览 {#1概览}

本文将带你了解如何在 Spring Boot 应用中动态地启动和停止 Kafka Listener。

2、依赖 {#2依赖}

首先，添加 spring-kafka 依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>3.1.2</version>
</dependency>

3、配置 Kafka 消费者 {#3配置-kafka-消费者}

生产者是向 Kafka Topic 发布（写入）事件的应用。

在本教程中，我们使用单元测试来模拟生产者向 Kafka Topic 发送事件。订阅 Topic 并处理事件流的消费者由应用中的 Listener（监听器）表示。该 Listener 被配置为处理来自 Kafka 的传入消息。

通过 KafkaConsumerConfig 类来配置 Kafka 消费者，其中包括 Kafka broker 的地址、消费者组 ID 以及 Key 和 Value 的反序列化器：

@Bean
public DefaultKafkaConsumerFactory<String, UserEvent> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.baeldung.spring.kafka.start.stop.consumer");
    return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(UserEvent.class));
}

4、配置 Kafka Listener {#4配置-kafka-listener}

在 Spring Kafka 中，用 @KafkaListener 对方法进行注解，就能创建一个 Listener，用于接收来自指定 Topic 的消息。

声明一个 UserEventListener 类：

@KafkaListener(id = Constants.LISTENER_ID, topics = Constants.MULTI_PARTITION_TOPIC, groupId = "test-group",
  containerFactory = "kafkaListenerContainerFactory", autoStartup = "false")
public void processUserEvent(UserEvent userEvent) {
    logger.info("Received UserEvent: " + userEvent.getUserEventId());
    userEventStore.addUserEvent(userEvent);
}

上述 Listener 等待来自 MULTI_PARTITION_TOPIC 的消息，并使用 processUserEvent() 方法处理这些消息。我们将 groupId 指定为 test-group，让消费者成为消费组的实例之一，从而促进跨多个实例的分布式处理。

使用 id 属性为每个 Listener 分配一个唯一标识符。在本例中，分配的 Listener ID 是 listener-id-1。

通过 autoStartup 属性，可以控制 Listener 是否在应用初始化时启动。在本例中，我们将其设置为 false，这意味着当应用启动时，Listener 不会自动启动。这种配置为我们提供了手动启动 Listener 的灵活性。

这种手动启动可由各种事件触发，如新用户注册、应用内的特定条件（如达到一定的数据量阈值）或管理操作（如通过管理界面手动启动 Listener）。例如，如果在线零售应用检测到秒杀抢购期间流量激增，它就会自动启动额外的 Listener 来处理增加的负载，从而优化性能。

UserEventStore 用于临时存储 Listener 接收到的事件：

@Component
public class UserEventStore {

    private final List<UserEvent> userEvents = new ArrayList<>();

    public void addUserEvent(UserEvent userEvent) {
        userEvents.add(userEvent);
    }

    public List<UserEvent> getUserEvents() {
        return userEvents;
    }

    public void clearUserEvents() {
        this.userEvents.clear();
    }
}

5、动态控制 Listener {#5动态控制-listener}

创建一个 KafkaListenerControlService，使用 KafkaListenerEndpointRegistry 动态地启动和停止 Kafka Listener：

@Service
public class KafkaListenerControlService {

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    public void startListener(String listenerId) {
        MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
        if (listenerContainer != null && !listenerContainer.isRunning()) {
            listenerContainer.start();
        }
    }

    public void stopListener(String listenerId) {
        MessageListenerContainer listenerContainer = registry.getListenerContainer(listenerId);
        if (listenerContainer != null && listenerContainer.isRunning()) {
            listenerContainer.stop();
        }
    }
}

KafkaListenerControlService 可以根据分配的 ID 精确地管理单个 Listener 实例。startListener() 和 stopListener() 方法都使用 listenerId 作为参数，允许我们根据需要启动和停止 Topic 的消息消费。

KafkaListenerEndpointRegistry 是在 Spring Application Context 中定义的所有 Kafka Listener 端点的中央 Repository。它监控这些 Listener 容器，从而允许对它们的状态（启动、停止或暂停）进行编程式控制。对于需要实时调整消息处理活动而无需重启整个应用的应序来说，这一功能至关重要。

6、验证动态控制 {#6验证动态控制}

接下来，在 Spring Boot 应用中测试 Kafka Listener 的动态启动和停止功能。

首先，启动 Listener：

kafkaListenerControlService.startListener(Constants.LISTENER_ID);

然后，通过发送并处理一个测试事件来验证 Listener 是否已激活：

UserEvent startUserEventTest = new UserEvent(UUID.randomUUID().toString()); 
producer.send(new ProducerRecord<>(Constants.MULTI_PARTITION_TOPIC, startUserEventTest)); 
await().untilAsserted(() -> assertEquals(1, this.userEventStore.getUserEvents().size())); 
this.userEventStore.clearUserEvents();

现在 Listener 已激活，我们批量发送十条消息进行处理。发送四条信息后，停止 Listener，然后将剩余的信息发送到 Kafka Topic：

for (long count = 1; count <= 10; count++) {
    UserEvent userEvent = new UserEvent(UUID.randomUUID().toString());
    Future<RecordMetadata> future = producer.send(new ProducerRecord<>(Constants.MULTI_PARTITION_TOPIC, userEvent));
    RecordMetadata metadata = future.get();
    if (count == 4) {
        await().untilAsserted(() -> assertEquals(4, this.userEventStore.getUserEvents().size()));
        this.kafkaListenerControlService.stopListener(Constants.LISTENER_ID);
        this.userEventStore.clearUserEvents();
    }
    logger.info("User Event ID: " + userEvent.getUserEventId() + ", Partition : " + metadata.partition());
}

在启动 Listener 之前，要确认 event store 中没有任何消息：

assertEquals(0, this.userEventStore.getUserEvents().size());
kafkaListenerControlService.startListener(Constants.LISTENER_ID);
await().untilAsserted(() -> assertEquals(6, this.userEventStore.getUserEvents().size()));
kafkaListenerControlService.stopListener(Constants.LISTENER_ID);

一旦 Listener 再次启动，它就会处理我们在 Listener 停止后发送到 Kafka Topic 的其余六条消息。

这个测试展示了 Spring Boot 应用动态管理 Kafka Listener 的能力。

7、使用案例 {#7使用案例}

动态 Listener 管理在需要高度适应性的场景中表现出色。例如，在负载高峰期，可以动态地启动额外的 Listener，以提高吞吐量并减少处理时间。相反，在维护或低流量期间，可以停止 Listener 以节省资源。这种灵活性还有利于在 "Feature flags" 后部署新功能，从而在不影响整个系统的情况下进行无缝的即时调整。

考虑这样一个场景：一个电子商务平台推出了一个新的推荐引擎，旨在通过根据浏览历史和购买模式推荐产品来提升用户体验。为了在全面推出之前验证该功能的有效性，我们决定在 "Feature flags" 后部署该功能。

激活此 "Feature flag" 可启动 Kafka Listener。当终端用户与平台互动时，由 Kafka Listener 提供支持的推荐引擎会处理传入的用户活动数据流，生成个性化的产品推荐。

当停用 "Feature flag" 时，就会停止 Kafka Listener，平台会默认使用现有的推荐引擎。这样，无论新引擎处于哪个测试阶段，都能确保无缝的用户体验。

当该功能处于激活状态时，我们可以积极收集数据，监控性能指标，并对推荐引擎进行调整。我们可以在多次迭代中重复这种功能测试，直到达到预期效果。

通过这种迭代过程，动态 Listener 管理被证明是一种有价值的工具。它允许无缝引入新功能