如何多次消费Kafka Topic中的消息？求重复获取记录方案

这个问题其实是Kafka消费场景里的常见需求——要重复获取已消费过的消息，核心就是搞定**消费偏移量（offset）**的控制。默认情况下，Kafka消费者会自动提交offset，消费完消息后offset会更新到最新位置，下次消费就从新位置开始，所以看不到之前的消息。下面给你几种可行的实现方案和代码示例：

方案1：每次启动消费者时从头开始拉取 #

如果只是想每次启动消费者都能拿到Topic里的所有历史消息，可以通过两个配置实现：

• 设置auto.offset.reset=earliest：告诉消费者如果找不到该消费组的offset记录，就从头开始消费
• 使用动态的消费组ID：每次启动用不同的组ID，这样Kafka会认为是新的消费组，直接触发从头消费

Java代码示例 #

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class RepeatConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        // 替换成你的Kafka集群地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 动态生成消费组ID，确保每次启动都是新组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "repeat-consumer-" + System.currentTimeMillis());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 找不到offset时从头消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 关闭自动提交，避免消费后自动更新offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // 替换成你要消费的Topic名称
            consumer.subscribe(Collections.singletonList("your-test-topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                records.forEach(record -> {
                    System.out.printf("拿到消息：offset=%d, key=%s, value=%s%n",
                            record.offset(), record.key(), record.value());
                });
                // 这里不提交offset，确保下次重启还能从头消费
            }
        }
    }
}

方案2：手动重置offset到指定位置 #

如果想用同一个消费组重复消费，可以在消费前手动把offset重置到起始位置（或某个具体offset）。这种方式适合需要反复消费某段消息的场景。

Java代码示例 #

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class ResetOffsetConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 固定消费组ID
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "fixed-repeat-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("your-test-topic"));
            
            // 先poll一次让消费者获取分区分配信息
            consumer.poll(Duration.ofMillis(100));
            // 重置所有分配到的分区的offset到起始位置
            consumer.seekToBeginning(consumer.assignment());

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                records.forEach(record -> {
                    System.out.printf("拿到消息：offset=%d, key=%s, value=%s%n",
                            record.offset(), record.key(), record.value());
                });

                // 如果需要每次循环都重新消费这段消息，可以在这里再次调用重置方法
                // consumer.seekToBeginning(consumer.assignment());
            }
        }
    }
}

如果只想重置到某个具体的offset，可以用consumer.seek(partition, targetOffset)，比如：

// 假设要重置到offset=0
consumer.assignment().forEach(partition -> consumer.seek(partition, 0));

方案3：禁用自动提交，手动控制提交时机 #

如果不想每次都从头消费，而是可以按需重复消费某段消息，可以禁用自动提交，只在确认不需要再重复消费时才手动提交offset。这样只要不提交，重启消费者就会从上次未提交的位置开始消费。

Java代码示例 #

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class ManualCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "manual-commit-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 禁用自动提交offset
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("your-test-topic"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                records.forEach(record -> {
                    System.out.printf("拿到消息：offset=%d, key=%s, value=%s%n",
                            record.offset(), record.key(), record.value());
                    // 业务处理逻辑...
                });

                // 只有当你确定不需要再重复消费这些消息时，才手动提交offset
                // consumer.commitSync();
            }
        }
    }
}

注意事项 #

• 如果你用的是同一个消费组，Kafka会在集群中保存该组的offset信息，所以重置offset或换组ID是关键。
• 生产环境中，重复消费可能会带来幂等性问题，要确保你的业务逻辑支持重复处理（比如重复插入数据不会导致重复记录）。

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