Spring Kafka：主题为空时解锁CountDownLatch及获取主题大小

你遇到的这个问题确实很常见——当Kafka指定分区里没有任何消息时，@KafkaListener的消费方法根本不会被触发，导致CountDownLatch一直处于等待状态，程序可能会卡死在这里。我给你分享几个实用的解决方案，你可以根据自己的业务场景来选：

方案一：主动查询主题是否为空 #

可以在程序启动后，通过Kafka的AdminClient主动检查目标分区的末尾偏移量，如果偏移量为0（说明分区里从来没存过消息），或者当前消费位置等于末尾偏移量（说明所有消息都已被消费完，当前为空），直接调用latch.countDown()。

代码示例如下：

@Autowired
private KafkaAdmin kafkaAdmin;

// 可以在@PostConstruct里调用，或者在等待latch前执行
private void checkEmptyTopic() {
    try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
        TopicPartition targetPartition = new TopicPartition(KAFKA_TOPIC, 0);
        // 查询分区的末尾偏移量
        Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(
                ListOffsetsResult.forTopicPartition(targetPartition)
        ).all().get();
        
        Long endOffset = endOffsets.get(targetPartition);
        // 如果末尾偏移量为0，说明分区为空
        if (endOffset == 0) {
            latch.countDown();
            log.info("检测到Kafka主题为空，已解锁CountDownLatch");
        }
    } catch (InterruptedException | ExecutionException e) {
        Thread.currentThread().interrupt();
        log.error("检查Kafka主题状态时出错", e);
    }
}

这个方案简单直接，适合一次性的消费任务（比如数据初始化、批量导出），但要注意同步查询会阻塞当前线程，要考虑对启动速度的影响。

方案二：通过容器自定义逻辑检测 #

可以自定义KafkaListenerContainerFactory，在消费者容器启动后，立即检查当前分区的消费位置和末尾偏移量，如果两者相等，说明没有可消费的消息，直接解锁latch。

代码示例：

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> factory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> consumerFactory
) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, consumerFactory);
    
    // 自定义容器初始化后的逻辑
    factory.setContainerCustomizer(container -> {
        container.start();
        Set<TopicPartition> assignedPartitions = container.getAssignedPartitions();
        if (!assignedPartitions.isEmpty()) {
            TopicPartition partition = assignedPartitions.iterator().next();
            Consumer<?, ?> consumer = container.getContainerProperties().getConsumer();
            
            try {
                Long currentOffset = consumer.position(partition);
                Long endOffset = consumer.endOffsets(Collections.singleton(partition)).get(partition);
                
                if (currentOffset.equals(endOffset)) {
                    latch.countDown();
                    log.info("容器启动后检测到无消息可消费，已解锁CountDownLatch");
                }
            } catch (WakeupException | InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("检测分区偏移量时出错", e);
            }
        }
    });
    
    return factory;
}

这个方案把检测逻辑和容器生命周期绑定，不需要额外的定时任务或者启动后检查，适合长期运行的消费者场景。

方案三：监听消费者空闲事件 #

Spring Kafka提供了消费者空闲事件的机制——当消费者在指定时间内没有收到任何消息时，会触发ListenerContainerIdleEvent。我们可以监听这个事件，判断是否是因为主题为空导致的空闲，进而解锁latch。

首先给容器工厂配置空闲超时时间：

@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> factory(
        ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
        ConsumerFactory<Object, Object> consumerFactory
) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, consumerFactory);
    
    // 设置空闲超时时间，比如5秒（根据实际业务调整）
    factory.getContainerProperties().setIdleEventInterval(5000);
    
    return factory;
}

然后编写事件监听器：

@EventListener
public void handleIdleConsumerEvent(ListenerContainerIdleEvent event) {
    Set<TopicPartition> assignedPartitions = event.getContainer().getAssignedPartitions();
    if (!assignedPartitions.isEmpty()) {
        TopicPartition partition = assignedPartitions.iterator().next();
        Consumer<?, ?> consumer = event.getConsumer();
        
        try {
            Long currentOffset = consumer.position(partition);
            Long endOffset = consumer.endOffsets(Collections.singleton(partition)).get(partition);
            
            // 当前消费位置等于末尾偏移量，说明没有新消息可消费
            if (currentOffset.equals(endOffset)) {
                latch.countDown();
                log.info("消费者空闲且检测到主题为空，已解锁CountDownLatch");
                // 可以关闭容器，避免重复触发空闲事件
                event.getContainer().stop();
            }
        } catch (WakeupException | InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("处理空闲事件时出错", e);
        }
    }
}

这个方案是事件驱动的方式，比较优雅，不需要主动轮询，适合对实时性要求不高的场景。你只需要调整好空闲超时时间，避免因为临时网络波动误判主题为空。

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