Spring Boot中Kafka消费者如何按分区从最新Offset开始消费

核心问题分析 #

你之前的错误在于在onMessage方法中调用seek——这个方法是消费者收到消息后才触发的，此时已经消费了旧消息，再调整偏移量为时已晚。正确的时机是在分区刚分配给消费者时就设置起始偏移量，也就是重写onPartitionsAssigned方法。

具体实现步骤 #

1. 存储目标偏移量 #

先把你已获取的各分区最新Offset保存起来（可以硬编码测试，也可以动态获取）：

public class OffsetConsumer implements AcknowledgingMessageListener<Integer, String>, ConsumerSeekAware {

    // 硬编码已获取的topic-99各分区最新Offset（测试用）
    private final Map<Integer, Long> targetOffsets = Map.of(
        0, 13192L,
        1, 12183L,
        2, 13178L,
        3, 13661L,
        4, 14026L
    );

    // 省略其他字段...

2. 重写onPartitionsAssigned方法设置偏移量 #

在分区分配给消费者时，直接调用seek跳转到指定Offset，这样消费者会从该Offset之后开始消费新数据：

@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> map, ConsumerSeekCallback consumerSeekCallback) {
    // 遍历所有分配到的分区
    map.keySet().forEach(partition -> {
        String topic = partition.topic();
        int partitionNum = partition.partition();
        // 获取当前分区对应的目标Offset
        Long targetOffset = targetOffsets.get(partitionNum);
        if (targetOffset != null) {
            // 设置消费者从指定Offset开始消费
            consumerSeekCallback.seek(topic, partitionNum, targetOffset);
        }
    });
}

3. （可选）动态获取最新Offset #

如果不想硬编码Offset，可以在监听器初始化时自动获取最新的endOffsets，替代硬编码的Map：

@Autowired
private KafkaConsumerConfig kafkaConsumerConfig;

private Map<Integer, Long> targetOffsets;

// 初始化时获取topic-99的最新Offset
@PostConstruct
public void initTargetOffsets() throws Exception {
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaConsumerConfig.consumerConfigs());
    consumer.subscribe(Collections.singletonList("topic-99"));
    
    Set<TopicPartition> assignment;
    while ((assignment = consumer.assignment()).isEmpty()) {
        consumer.poll(Duration.ofMillis(500));
    }
    
    // 获取各分区的最新Offset
    Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assignment);
    // 转换为分区号到Offset的映射
    targetOffsets = endOffsets.entrySet().stream()
        .collect(Collectors.toMap(entry -> entry.getKey().partition(), Map.Entry::getValue));
    
    consumer.close();
}

4. 清理无效代码 #

删除onMessage中错误的seek调用，修改后的onMessage只需处理业务逻辑：

@Override
public void onMessage(ConsumerRecord<Integer, String> record, Acknowledgment acknowledgment) {
    // 处理新消息的业务逻辑
    System.out.println("收到新消息：" + record.value());
    // 手动提交偏移量（如果配置了手动提交）
    acknowledgment.acknowledge();
}

关键说明 #

• onPartitionsAssigned方法是消费者分配到分区后立即触发的，此时还未开始消费消息，在这里设置seek能确保消费者直接跳转到指定位置，不会消费旧数据。
• 如果你的消费者配置了自动提交偏移量，确保在消费新消息后正确提交，避免重启后重复消费。

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