使用Spring Kafka/Spring Boot，@KafkaListener如何每次启动从头消费主题？

刚好之前做过类似的需求，用@KafkaListener实现每次应用启动都从主题起始位置（offset=0）读取消息，有几种实用的方案，帮你梳理清楚：

这是最可靠的方式，利用Kafka的重平衡机制，在消费者分配到分区时强制重置offset到0。不管你的消费组之前有没有留存的offset记录，每次应用启动（或者重平衡发生时）都会触发重置操作。

步骤1：实现自定义重平衡监听器 #

创建一个类实现ConsumerAwareRebalanceListener，在分区分配时执行offset重置：

@Component
public class StartupOffsetResetListener implements ConsumerAwareRebalanceListener {

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // 遍历所有分配到的分区，将offset重置为0
        partitions.forEach(partition -> consumer.seek(partition, 0));
    }

    @Override
    public void onPartitionsRevoked(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        // 这里可以做一些清理操作，比如提交当前offset（可选）
    }
}

步骤2：在@KafkaListener中引用该监听器 #

把自定义的重平衡监听器注入到你的消息监听方法中：

@Autowired
private StartupOffsetResetListener resetListener;

@KafkaListener(
    topics = "your_target_topic",
    groupId = "your_consumer_group",
    rebalanceListener = "#{resetListener}"
)
public void handleMessage(String message, ConsumerRecord<?, ?> record) {
    // 你的消息处理逻辑
    System.out.println("Received message from offset 0+: " + message);
}

这个方案的优势在于：逻辑清晰，不依赖消费组的历史状态，每次启动都能确保从最开始读取，非常适合需要重复消费全量数据的场景。

如果你的应用是单实例运行，或者只是临时需要全量消费，可以通过动态生成group.id的方式配合auto.offset.reset=earliest实现。因为Kafka会把新的消费组视为没有历史offset，此时auto.offset.reset会生效，从起始位置开始消费。

示例代码：

@KafkaListener(
    topics = "your_target_topic",
    properties = {
        "auto.offset.reset=earliest",
        "group.id=temp_group_#{T(System).currentTimeMillis()}"
    }
)
public void handleMessage(String message) {
    // 消息处理逻辑
}

⚠️ 注意：这个方案的弊端很明显——每次启动都会生成新的消费组，Kafka会留存大量无用的消费组offset记录；如果是多实例部署，每个实例的消费组ID不同，会导致同一条消息被多个实例重复消费，所以只适合测试或者一次性的离线任务。

通过监听应用启动事件，手动控制消费者容器的生命周期，重置offset后再启动容器。这种方式适合需要全局统一控制的场景，但逻辑相对复杂一些。

示例代码：

@Autowired
private KafkaListenerEndpointRegistry listenerRegistry;

@EventListener(ApplicationStartedEvent.class)
public void resetOffsetOnStartup() {
    // 遍历所有Kafka监听容器
    listenerRegistry.getListenerContainers().forEach(container -> {
        // 先停止容器，避免消费过程中重置offset引发问题
        container.stop();
        
        // 创建临时消费者，用于获取分区并重置offset
        ConsumerFactory<?, ?> consumerFactory = container.getConsumerFactory();
        try (Consumer<?, ?> consumer = consumerFactory.createConsumer("your_consumer_group", null)) {
            consumer.subscribe(Collections.singletonList("your_target_topic"));
            // 获取分配的分区集合
            Set<TopicPartition> partitions = consumer.assignment();
            // 等待分区分配完成（避免空分区）
            consumer.poll(Duration.ofSeconds(1));
            partitions = consumer.assignment();
            
            // 重置每个分区的offset到0
            partitions.forEach(partition -> consumer.seek(partition, 0));
        }
        
        // 重启容器，开始消费
        container.start();
    });
}

这个方案需要注意容器的启停顺序，以及分区分配的等待时间，避免出现空分区导致重置失败的情况。

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