Kafka消费者不从开头消费

以下是一个示例代码，展示了如何创建一个Kafka消费者，并从指定的偏移量开始消费消息：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

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

public class KafkaConsumerExample {

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String topic = "my_topic";
        String groupId = "my_group_id";
        int partition = 0;
        long offset = 10; // 设置开始消费的偏移量

        // 创建消费者配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 分配给消费者指定分区和偏移量
        TopicPartition topicPartition = new TopicPartition(topic, partition);
        consumer.assign(Collections.singleton(topicPartition));
        consumer.seek(topicPartition, offset);

        // 开始消费消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received message: topic = %s, partition = %s, offset = %d, key = %s, value = %s\n",
                        record.topic(), record.partition(), record.offset(), record.key(), record.value());
            }
        }
    }
}

在上面的代码中，我们首先创建了一个Kafka消费者，并设置了所需的配置，如Kafka服务器地址、消费者组ID等。然后，我们使用assign方法将指定的主题分区和偏移量分配给消费者。最后，我们使用一个循环来持续消费消息，通过poll方法从Kafka获取消息，然后处理每条消息。

请注意，这里的消费者是手动指定偏移量进行消费的，而不是从最早或最新的偏移量开始。您可以根据需要调整offset变量的值，以指定从哪个偏移量开始消费消息。