如何实现Kafka Topic1与Topic2顺序消费且不停止Topic1监听器？

你的核心问题是ContainerGroupSequencer会直接停止topic1的容器组，不符合“保留topic1监听器持续运行”的需求。可以通过手动控制容器启停状态+偏移量校验的方式实现需求，具体步骤如下：

1. 调整监听器配置 #

给topic2的监听器设置autoStartup="false"，让它初始处于暂停状态；同时确保topic1的监听器保持默认自动启动：

@KafkaListener(id = "listen3", topics = "topic1", concurrency = "2")
public void listen1(String in) {
    // 处理topic1消息逻辑
}

@KafkaListener(id = "listen4", topics = "topic2", concurrency = "2", autoStartup = "false")
public void listen2(String in) {
    // 处理topic2消息逻辑
}

2. 注入容器注册表 #

通过KafkaListenerEndpointRegistry获取和控制两个监听器容器：

@Autowired
private KafkaListenerEndpointRegistry containerRegistry;

3. 实现topic1消费完成校验 #

通过对比topic1各分区的已消费偏移量和最新偏移量，判断是否所有消息都已消费：

private boolean isTopic1FullyConsumed() {
    Consumer<?, ?> consumer = containerRegistry.getListenerContainer("listen3")
            .getConsumerFactory().createConsumer();
    List<PartitionInfo> partitions = consumer.partitionsFor("topic1");
    
    // 无分区直接视为消费完成
    if (partitions == null || partitions.isEmpty()) {
        return true;
    }

    for (PartitionInfo partition : partitions) {
        TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
        // 获取分区最新偏移量
        long endOffset = consumer.endOffsets(Collections.singleton(topicPartition)).get(topicPartition);
        // 获取已提交的消费偏移量
        OffsetAndMetadata committedOffset = consumer.committed(topicPartition);
        
        // 若存在未消费完的分区，返回false
        if (committedOffset == null || committedOffset.offset() < endOffset) {
            return false;
        }
    }
    consumer.close();
    return true;
}

4. 定时检查并启动topic2容器 #

在项目启动后，定时校验topic1的消费状态，一旦确认消费完成，立即启动topic2的容器：

@PostConstruct
public void startSequencedConsumption() {
    // 启动定时任务，每5秒检查一次topic1消费状态
    ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
    scheduler.scheduleAtFixedRate(() -> {
        if (isTopic1FullyConsumed()) {
            // 启动topic2的监听器容器
            containerRegistry.getListenerContainer("listen4").start();
            // 停止定时检查任务
            scheduler.shutdown();
        }
    }, 0, 5, TimeUnit.SECONDS);
}

关键说明 #

• 这种方式下，topic1的监听器会持续运行，后续新产生的消息依然会被消费；
• 定时检查的间隔可以根据业务场景调整，避免过于频繁影响性能；
• 如果需要更精准的消费完成判断，可以结合Kafka的消费提交事件，在每次提交偏移量后触发校验。

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