Kafka将内部主题的偏移存储在哪里？

Kafka将内部主题的偏移存储在Zookeeper或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.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class OffsetStorageExample {
    public static void main(String[] args) {
        // Kafka broker地址
        String bootstrapServers = "localhost:9092";
        // 消费者组ID
        String groupId = "my-consumer-group";
        // 要订阅的主题
        String topic = "my-topic";

        // 配置消费者属性
        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");

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

        // 获取主题的分区信息
        List<PartitionInfo> partitions = consumer.partitionsFor(topic);

        // 订阅所有分区
        List<TopicPartition> topicPartitions = new ArrayList<>();
        for (PartitionInfo partition : partitions) {
            topicPartitions.add(new TopicPartition(partition.topic(), partition.partition()));
        }
        consumer.assign(topicPartitions);

        // 读取存储的偏移量
        Map<TopicPartition, Long> offsets = new HashMap<>();
        for (TopicPartition topicPartition : topicPartitions) {
            // 从Kafka内部主题中读取偏移量
            long offset = consumer.position(topicPartition);
            offsets.put(topicPartition, offset);
        }

        // 打印偏移量
        for (Map.Entry<TopicPartition, Long> entry : offsets.entrySet()) {
            System.out.println("Partition: " + entry.getKey() + ", Offset: " + entry.getValue());
        }

        // 关闭消费者
        consumer.close();
    }
}

上述代码使用KafkaConsumer API来订阅主题，并使用position()方法从Kafka内部主题中读取偏移量。最后，打印出每个分区的偏移量。

请注意，在实际生产环境中，通常会使用更复杂的逻辑来处理偏移量，例如将偏移量存储在外部数据库中，或使用Kafka的__consumer_offsets内部主题来存储偏移量。