Spring Kafka单消费者接收随机分区消息及偏移量重跑问题

问题场景 #

我们使用单个Spring Kafka消费者，通过@KafkaListener监听配置的topic，代码如下：

@KafkaListener(topics = "${kafka.topic.mytopic}")
public void consume(@Payload String message,
                    @Header(KafkaHeaders.OFFSET) Long offset,
                    @Header(KafkaHeaders.RECEIVED_PARTITION) int partition) {
   log.info("Kafka partition:{} offset:{} got an event:{} ", partition, offset, message);
}

实际输出示例：

Kafka partition:1 offset:963582 got an event:xxx
Kafka partition:1 offset:963583 got an event:xxx
Kafka partition:2 offset:993582 got an event:xxx
Kafka partition:0 offset:935482 got an event:xxx

发送4条消息后，消费者收到4条消息，但每条消息来自随机分区。我们预期要么消费者仅监听一个分区，要么若监听全部5个分区，会收到每条消息的4条重复数据。此外，若服务发生故障，我们希望从指定offset重新消费，但目前不清楚分区数量且消息乱序，该如何实现？

问题1：消费行为不符合预期的原因 #

你的预期存在对Kafka消费者机制的误解：

• Kafka消费者分组规则：同一消费者组内的消费者会分摊topic的分区，如果组内只有单个消费者，它会被分配到该topic的所有分区。
• 消息分配逻辑：默认情况下，生产者发送消息时会按轮询（无key）或哈希key的方式将消息分发到不同分区，所以发送4条消息会被分到不同分区，单个消费者会从每个分区各消费一条，不会出现重复。
• 重复消费的场景：只有当多个消费者属于不同的消费者组时，才会各自收到全部分区的消息；同一组内的多个消费者只会分摊分区，不会重复消费同一条消息。

问题2：从指定offset重新消费的实现方案 #

步骤1：获取topic的所有分区 #

可以通过Spring Kafka提供的AdminClient获取目标topic的分区列表，无需提前知道分区数量：

@Autowired
private KafkaAdmin kafkaAdmin;

public Set<TopicPartition> getTopicPartitions(String topic) {
    try (AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties())) {
        DescribeTopicsResult result = adminClient.describeTopics(Collections.singleton(topic));
        TopicDescription description = result.values().get(topic).get();
        return description.partitions().stream()
                .map(p -> new TopicPartition(topic, p.partition()))
                .collect(Collectors.toSet());
    } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException("Failed to get topic partitions", e);
    }
}

步骤2：手动指定offset消费 #

有两种常用实现方式：

方式一：实现ConsumerSeekAware接口

在消费者类中实现该接口，在消费者初始化时指定每个分区的目标offset：

@Component
public class KafkaMessageConsumer implements ConsumerSeekAware {
    private final String targetTopic = "${kafka.topic.mytopic}";
    private final long targetOffset = 1000L; // 自定义指定的offset

    @Autowired
    private KafkaAdmin kafkaAdmin;

    @KafkaListener(topics = "${kafka.topic.mytopic}")
    public void consume(@Payload String message,
                        @Header(KafkaHeaders.OFFSET) Long offset,
                        @Header(KafkaHeaders.RECEIVED_PARTITION) int partition) {
        log.info("Kafka partition:{} offset:{} got an event:{} ", partition, offset, message);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        Set<TopicPartition> partitions = getTopicPartitions(targetTopic);
        // 为每个分区设置指定offset
        partitions.forEach(partition -> callback.seek(partition.topic(), partition.partition(), targetOffset));
    }
}

方式二：通过KafkaConsumer手动seek

如果需要更灵活的控制，可直接创建KafkaConsumer实例执行seek操作：

@Autowired
private ConsumerFactory<String, String> consumerFactory;

public void seekToSpecificOffset(String topic, long targetOffset) {
    try (KafkaConsumer<String, String> consumer = consumerFactory.createConsumer()) {
        Set<TopicPartition> partitions = getTopicPartitions(topic);
        consumer.assign(partitions);
        partitions.forEach(partition -> consumer.seek(partition, targetOffset));
        // 后续可手动拉取消息，或交由@KafkaListener接管消费逻辑
    }
}

消息乱序的处理 #

• Kafka仅保证单个分区内的消息有序，跨分区的消息消费顺序是随机的（消费者会轮询拉取不同分区的消息）。
• 若需全局消息有序，只能让所有消息发送到同一个分区：可通过指定相同的消息key，或自定义分区器将所有消息路由到同一分区，但这会牺牲Kafka的并行处理能力。
• 若无需全局有序，可基于分区+offset做幂等处理，避免重复消费带来的业务问题。

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