如何用Spring Boot每秒从头批量消费Kafka主题全部消息？

要实现每秒重复从头批量拉取并打印主题全部消息，你需要放弃常规的@KafkaListener持续监听模式，改用定时任务+手动控制消费者offset的方式，具体步骤如下：

1. 配置Kafka消费者参数 #

在application.yml中添加批量消费相关配置：

spring:
  kafka:
    consumer:
      bootstrap-servers: your-kafka-broker:9092  # 替换为你的Kafka服务地址
      group-id: batch-consumer-group  # 固定group即可，因为每次都会主动重置offset
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      max-poll-records: 1000  # 每次批量拉取的最大消息数，可根据实际需求调整

2. 编写定时批量消费服务 #

创建服务类，用@Scheduled每秒触发一次拉取操作，每次拉取前主动将offset重置到主题开头：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.Collections;

@Service
public class BatchKafkaConsumerService {

    private final ConsumerFactory<String, String> consumerFactory;
    private static final String TARGET_TOPIC = "retry-events";

    @Autowired
    public BatchKafkaConsumerService(ConsumerFactory<String, String> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    @Scheduled(fixedRate = 1000)  // 每秒执行一次消费任务
    public void consumeBatchFromStart() {
        // 每次创建新消费者，避免offset持久化影响下次拉取
        try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
            // 订阅目标主题
            consumer.subscribe(Collections.singletonList(TARGET_TOPIC));
            
            // 触发订阅并获取分区信息，重置offset到主题最开始位置
            consumer.poll(Duration.ofMillis(100));
            for (TopicPartition partition : consumer.assignment()) {
                consumer.seekToBeginning(Collections.singletonList(partition));
            }
            
            // 批量拉取消息，超时1秒（无新消息时立即返回）
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            
            // 打印批量消息
            if (!records.isEmpty()) {
                System.out.printf("===== 本次拉取到 %d 条消息 =====%n", records.count());
                records.forEach(record -> {
                    System.out.printf("Received message: %s (offset: %d)%n", record.value(), record.offset());
                });
            } else {
                System.out.println("本次未拉取到消息");
            }

            // 无需提交offset，因为每次都会从头消费
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3. 开启定时任务 #

在Spring Boot启动类上添加@EnableScheduling注解，启用定时任务功能：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
public class YourApplication {
    public static void main(String[] args) {
        SpringApplication.run(YourApplication.class, args);
    }
}

原有代码不生效的原因 #

• 常规@KafkaListener是持续长连接消费，一旦消费过的offset被记录（即使enable-auto-commit=false，Spring Kafka默认也可能在消费完成后提交offset），后续会从上次的offset位置继续消费，不会自动从头。
• 你需要的是周期性从头拉取，必须主动重置offset到主题开头，且用定时任务触发单次拉取，而非持续监听。

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