Kafka 0.9版本之前，offset存储在zookeeper，0.9版本之后，默认将offset存储在Kafka的一个内置的topic中。除此之外，Kafka还可以选择自定义存储offset。
    offset的维护是相当繁琐的，因为需要考虑到消费者的Rebalace。
    当有新的消费者加入消费者组、已有的消费者推出消费者组或者所订阅的主题的分区发生变化，就会触发到分区的重新分配，重新分配的过程叫做Rebalance。
    消费者发生Rebalance之后，每个消费者消费的分区就会发生变化。因此消费者要首先获取到自己被重新分配到的分区，并且定位到每个分区最近提交的offset位置继续消费。
    要实现自定义存储offset，需要借助ConsumerRebalanceListener，以下为示例代码，其中提交和获取offset的方法，需要根据所选的offset存储系统自行实现。

    1. package com.consumer;
    3. import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
    4. import org.apache.kafka.clients.consumer.ConsumerRecord;
    5. import org.apache.kafka.clients.consumer.ConsumerRecords;
    6. import org.apache.kafka.clients.consumer.KafkaConsumer;
    7. import org.apache.kafka.common.TopicPartition;
    9. import java.util.*;
    11. public class CustomConsumer4 {
    12.     // 记录分区的信息,这个也可以用redis等等存储. key是主体的名字和分区号
    14.     private static Map<TopicPartition, Long> currentOffset = new HashMap<>();
    16.     public static void main(String[] args) {
    18. //创建配置信息
    19.         Properties props = new Properties();
    21. //Kafka集群
    22.         props.put("bootstrap.servers", "zjj101:9092");
    24. //消费者组，只要group.id相同，就属于同一个消费者组
    25.         props.put("group.id", "test");
    27. //关闭自动提交offset
    28.         props.put("enable.auto.commit", "false");
    30.         //Key和Value的反序列化类
    31.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    32.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    34.         //创建一个消费者
    35.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    37.         //消费者订阅主题
    38.         consumer.subscribe(Arrays.asList("first"), new ConsumerRebalanceListener() {
    40.             /*该方法会在分区之前调用
    41.             分区之前各个消费者应该把offset提交*/
    42.             @Override
    43.             public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    44.                 // 提交旧的offset
    45.                 commitOffset(currentOffset);
    46.             }
    48.             /*该方法会在分区之后调用
    49.             分区之后应该给你offset信息重新给你*/
    50.             @Override
    51.             public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    52.                 //获取新的offset
    53.                 currentOffset.clear();
    54.                 for (TopicPartition partition : partitions) {
    55.                     consumer.seek(partition, getOffset(partition));//定位到最近提交的offset位置继续消费
    56.                 }
    57.             }
    58.         });
    60.         while (true) {
    61.             ConsumerRecords<String, String> records = consumer.poll(100);//消费者拉取数据
    62.             for (ConsumerRecord<String, String> record : records) {
    63.                 System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    64.                 currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
    65.             }
    66.             commitOffset(currentOffset);//异步提交
    67.         }
    68.     }
    70.     //获取某分区的最新offset
    71.     private static long getOffset(TopicPartition partition) {
    72.         return 0;
    73.     }
    75.     //提交该消费者所有分区的offset
    76.     private static void commitOffset(Map<TopicPartition, Long> currentOffset) {
    78.     }
    79. }