如何在Compacted Topic2的KafkaListener加载完数据后启动Topic1的KafkaListener

核心逻辑是通过Spring Kafka提供的监听器生命周期控制能力，先禁用topic1监听器的自动启动，待topic2全量数据读取、缓存构建完成后再手动启动topic1的监听器，具体实现可参考以下两种方案：

方案1：手动控制监听器启动状态（推荐） #

步骤1：关闭topic1监听器的自动启动 #

给@KafkaListener注解添加autoStartup = "false"属性，避免服务启动时自动开始消费topic1：

@KafkaListener(id = "topic1Listener", topics = "topic1", autoStartup = "false")
public void consumeTopic1(String message) {
    // 基于缓存的业务处理逻辑
}

步骤2：判定topic2全量读取完成的时机 #

针对compacted topic从头读取的场景，可通过Spring Kafka提供的ConsumerPartitionEndEvent事件判定全量加载完成：当所有分配给当前实例的topic2分区都触发该事件时，说明全量数据已经读取完毕。

步骤3：加载完成后手动启动topic1监听器 #

import org.springframework.kafka.event.ConsumerPartitionEndEvent;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import java.util.concurrent.atomic.AtomicInteger;

// 提前通过Kafka Admin获取topic2的分区总数，或监听ConsumerPartitionAssignmentEvent获取当前实例分配到的分区数
private int totalPartitionNum;
private AtomicInteger finishedPartitionNum = new AtomicInteger(0);

@KafkaListener(id = "topic2Listener", topics = "topic2", properties = {"auto.offset.reset=earliest"})
public void consumeTopic2(String message) {
    // 写入内存缓存的逻辑
}

@EventListener
public void onPartitionReadFinished(ConsumerPartitionEndEvent event) {
    // 过滤仅处理topic2的分区事件
    boolean isTopic2Partition = event.getTopicPartitions()
            .stream()
            .anyMatch(tp -> "topic2".equals(tp.topic()));
    if (!isTopic2Partition) {
        return;
    }
    int currentFinished = finishedPartitionNum.incrementAndGet();
    // 所有分区都读取到末尾，全量加载完成
    if (currentFinished == totalPartitionNum) {
        KafkaListenerEndpointRegistry registry = applicationContext.getBean(KafkaListenerEndpointRegistry.class);
        // 启动topic1的监听器
        registry.getListenerContainer("topic1Listener").start();
    }
}

方案2：依赖顺序控制（适合topic2数据量小的场景） #

如果topic2数据量不大，可以直接在服务初始化阶段同步完成全量加载：

1. 封装独立的缓存加载类Topic2CacheLoader，在@PostConstruct方法中同步拉取topic2的全量数据，完成内存缓存构建
2. 给topic1监听器所在的类添加@DependsOn("topic2CacheLoader")注解，Spring会保证缓存加载Bean初始化完成后，才会初始化topic1的监听器

注意事项 #

• 集群部署场景下，每个实例需要单独加载topic2全量数据，注意给topic2的监听器配置独立的消费者组，避免分区被分配到不同实例导致全量数据加载不全
• 缓存全量加载完成后，可保留topic2的监听器持续消费后续增量消息，实时更新内存缓存，保证数据一致性
• 若topic2数据量极大，建议提前评估加载耗时，避免触发服务启动超时检测

内容的提问来源于stack exchange，提问作者user3215114