如何在Java SpringBoot中不消费消息统计Kafka消息数并按需触发消费

核心思路 #

通过Kafka AdminClient获取指定主题各分区的末尾偏移量与消费者组的当前消费偏移量，计算待消费消息总数；结合定时任务或启动时检查，当总数超过设定阈值时启动消费容器，否则保持暂停状态。全程无需消费消息即可完成统计。

具体实现步骤 #

1. 配置Kafka AdminClient #

在SpringBoot配置类中注入AdminClient实例，用于查询分区偏移量：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaAdminConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public AdminClient kafkaAdminClient() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        return AdminClient.create(configs);
    }
}

2. 实现消息数量统计工具类 #

编写工具类封装偏移量计算逻辑，统计待消费消息总数：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ExecutionException;

@Component
public class KafkaMessageCounter {

    @Autowired
    private AdminClient adminClient;

    /**
     * 计算指定主题、消费者组的待消费消息总数
     */
    public long calculatePendingMessages(String topic, String consumerGroup) throws ExecutionException, InterruptedException {
        // 1. 获取主题所有分区
        Collection<TopicPartition> partitions = adminClient.describeTopics(topic)
                .all().get().get(topic).partitions().stream()
                .map(partitionInfo -> new TopicPartition(topic, partitionInfo.partition()))
                .toList();

        // 2. 获取分区末尾偏移量
        Map<TopicPartition, Long> endOffsets = adminClient.listOffsets(Map.ofEntries(
                partitions.stream().map(p -> Map.entry(p, OffsetAndMetadata.latest()))
                        .toArray(Map.Entry[]::new)
        )).all().get();

        // 3. 获取消费者组当前消费偏移量
        DescribeConsumerGroupsResult groupResult = adminClient.describeConsumerGroups(consumerGroup);
        ConsumerGroupDescription groupDesc = groupResult.all().get().get(consumerGroup);
        Map<TopicPartition, OffsetAndMetadata> currentOffsets = adminClient.listConsumerGroupOffsets(consumerGroup)
                .partitionsToOffsetAndMetadata().get();

        // 4. 计算待消费总数
        long totalPending = 0;
        for (TopicPartition partition : partitions) {
            Long endOffset = endOffsets.get(partition);
            OffsetAndMetadata currentOffset = currentOffsets.getOrDefault(partition, new OffsetAndMetadata(0));
            if (endOffset != null) {
                totalPending += endOffset - currentOffset.offset();
            }
        }
        return totalPending;
    }
}

3. 配置可控制的Kafka消费容器 #

使用ConcurrentKafkaListenerContainerFactory配置消费容器，默认设置为暂停状态，后续通过代码控制启停：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(ConsumerFactory<?, ?> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<?, ?> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 初始状态暂停消费
        factory.setAutoStartup(false);
        return factory;
    }
}

4. 实现阈值触发的消费控制逻辑 #

通过定时任务定期检查消息数量，控制消费容器的启停：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumerController {

    @Autowired
    private KafkaMessageCounter messageCounter;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Value("${kafka.topic.target}")
    private String targetTopic;

    @Value("${kafka.consumer.group-id}")
    private String consumerGroup;

    @Value("${kafka.consumer.threshold:1000}")
    private long threshold;

    // 每隔30秒检查一次
    @Scheduled(fixedRate = 30000)
    public void checkAndControlConsumption() {
        try {
            long pendingCount = messageCounter.calculatePendingMessages(targetTopic, consumerGroup);
            String containerId = "kafkaListenerContainer"; // 对应@KafkaListener的id属性

            if (pendingCount >= threshold && !registry.getListenerContainer(containerId).isRunning()) {
                // 超过阈值，启动消费
                registry.getListenerContainer(containerId).start();
                System.out.printf("待消费消息数(%d)达到阈值，启动消费%n", pendingCount);
            } else if (pendingCount < threshold && registry.getListenerContainer(containerId).isRunning()) {
                // 低于阈值，暂停消费
                registry.getListenerContainer(containerId).pause();
                System.out.printf("待消费消息数(%d)低于阈值，暂停消费%n", pendingCount);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

5. 编写Kafka消息监听器 #

给监听器指定容器id，对应上面控制逻辑中的容器标识：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageListener {

    @KafkaListener(id = "kafkaListenerContainer", topics = "${kafka.topic.target}", groupId = "${kafka.consumer.group-id}")
    public void listen(String message) {
        // 业务消费逻辑
        System.out.println("消费消息: " + message);
    }
}

关键注意事项 #

• AdminClient资源管理：AdminClient是线程安全的，无需每次调用都创建实例，复用即可。
• 分区动态变化：如果主题分区数量动态调整，统计逻辑会自动获取最新分区，无需额外配置。
• 消费组偏移量准确性：确保统计时使用的消费者组与实际消费的组一致，否则偏移量计算会出错。
• 定时任务频率：根据业务场景调整检查间隔，避免过于频繁调用AdminClient增加Kafka集群压力。

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