private static final String TOPIC="HLJ_TOPIC_JAVA";
    private static final String BROKER_LIST="localhost:9092";
    private static final String GROUP_ID="HTEST_GROUP";

    private static KafkaConsumer<String,String> kafkaConsumer = null;

    static {
        Properties properties = initConfig();
        kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Collections.singletonList(TOPIC));
    }

    private static Properties initConfig(){
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BROKER_LIST);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,GROUP_ID);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
        return properties;
    }

@Test
public void consumer(){
  try {
    //注释1
    while (true) {
      //注释2
      ConsumerRecords<String, String> records = consumer.poll(100);
      //注释3
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", record.partition(), record.offset(), record.key(), record.value());
      }
    }
  } catch (Exception e) {
    log.error("消费者处理数据失败", e);
  } finally {
    //注解4
    consumer.close();
  }
}

public class KafkaConsumerThread  implements Runnable{

  private KafkaConsumer<String,String> consumer;
  private AtomicBoolean closed = new AtomicBoolean(false);
  public KafkaConsumerThread(){

  }
  // 构造方法 生成自己的consumer
  public KafkaConsumerThread(Properties props) {
    this.consumer = new KafkaConsumer<>(props);
  }

  @Override
  public void run() {
    try {
      // 消费同一主题
      consumer.subscribe(Collections.singletonList("six-topic"));
      // 线程名称
      String threadName = Thread.currentThread().getName();
      while (!closed.get()){
        ConsumerRecords<String, String> records = consumer.poll(3000);
        for (ConsumerRecord<String, String> record : records) {
          System.out.printf("Context: Thread-name= %s, topic= %s partition= %s, offset= %d, key= %s,value= %s\n",threadName,record.topic(),record.partition(),record.offset(),record.key(),record.value());
        }
      }
    }catch (WakeupException e){
      e.printStackTrace();
    }finally {
      consumer.close();
    }
  }

  /**
     * 关闭消费
     */
  public void shutdown(){
    closed.set(true);
    // wakeup 可以安全地从外部线程来中断活动操作
    consumer.wakeup();
  }

  public static void main(String[] args) {
    Properties props = new Properties();
    props.put("bootstrap.servers", "XXXXXXX:9093");
    props.put("group.id", "thread-1");//消费者组，只要group.id相同，就属于同一个消费者组
    props.put("enable.auto.commit", "true");//自动提交offset
    props.put("auto.offset.reset", "earliest");
    props.put("auto.commit.interval.ms", "1000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("session.timeout.ms", "30000");
    props.put("max.poll.records",6);
    // 运行三个线程，消费同一个topic 这个topic的分区必须大于等于3 否则会有消费者消费不到数据
    for (int i = 0; i < 3 ; i++) {
      new Thread(new KafkaConsumerThread(props),"Thread"+i).start();
    }
  }
}

public class KafkaConsumerThread2 implements Runnable {
  // 存储消息 先进先出队列
  private LinkedBlockingQueue<ConsumerRecords<String,String>> list;
  private AtomicBoolean closed = new AtomicBoolean(false);

  public KafkaConsumerThread2() {
  }

  public KafkaConsumerThread2(LinkedBlockingQueue<ConsumerRecords<String, String>> list) {
    this.list = list;
  }

  @Override
  public void run() {
    // 线程名称
    String threadName = Thread.currentThread().getName();
    // 处理消息
    while (!closed.get()){
      try {
        ConsumerRecords<String, String> records = list.take();
        System.out.println("消息数量"+records.count());
        if (records.isEmpty()){
          System.out.printf("队列为空，不消费数据，Thread-name= %s\n",threadName);
        }else {
          for (ConsumerRecord<String, String> record : records) {
            Thread.sleep(3000);
            System.out.printf("Context: Thread-name= %s, topic= %s partition= %s, offset= %d, key= %s,value= %s\n",threadName,record.topic(),record.partition(),record.offset(),record.key(),record.value());
          }
        }
      }catch (InterruptedException e){
        e.printStackTrace();
      }
    }
  }

  public static void main(String[] args) {
    LinkedBlockingQueue<ConsumerRecords<String, String>> list = new LinkedBlockingQueue<>();
    Properties props = new Properties();
    props.put("bootstrap.servers", "10.33.68.68:9093");
    props.put("group.id", "thread-5");//消费者组，只要group.id相同，就属于同一个消费者组
    props.put("enable.auto.commit", "true");//自动提交offset
    props.put("auto.offset.reset", "earliest");
    props.put("auto.commit.interval.ms", "1000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("session.timeout.ms", "30000");
    props.put("max.poll.records",5);
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    // 消费同一主题
    consumer.subscribe(Collections.singletonList("six-topic"));
    // 开启三个线程处理队列中的消息
    for (int i = 0; i <3 ; i++) {
      new Thread(new KafkaConsumerThread2(list),"thread-"+i).start();
    }
    while (true){
      ConsumerRecords<String, String> records = consumer.poll(1000);
      try {
        list.put(records);
        //Thread.sleep(3000);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
}

public class ConsumerDealThread implements Runnable{
  private ConsumerRecord record;

  public ConsumerDealThread(ConsumerRecord record) {
    this.record = record;
  }

  public void run() {
    try {
      Thread.sleep(2000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    System.out.printf("Context: Thread-name= %s, topic= %s partition= %s, offset= %d, key= %s,value= %s\n",Thread.currentThread().getName(),record.topic(),record.partition(),record.offset(),record.key(),record.value());
  }
}

public class KafkaConsumerThread3 {
  public static void main(String[] args) {
    LinkedBlockingQueue<ConsumerRecords<String, String>> list = new LinkedBlockingQueue<>();
    Properties props = new Properties();
    props.put("bootstrap.servers", "10.33.68.68:9093");
    props.put("group.id", "thread-18");//消费者组，只要group.id相同，就属于同一个消费者组
    props.put("enable.auto.commit", "true");//自动提交offset
    props.put("auto.offset.reset", "earliest");
    props.put("auto.commit.interval.ms", "1000");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("session.timeout.ms", "30000");
    props.put("max.poll.records",5);
    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    // 消费同一主题
    consumer.subscribe(Collections.singletonList("six-topic"));
    ExecutorService executor = new ThreadPoolExecutor(3, 3, 0, TimeUnit.MILLISECONDS,
                                                      new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy());
    while (true){
      ConsumerRecords<String, String> records = consumer.poll(1000);
      try {
        for (ConsumerRecord<String, String> record : records) {
          executor.submit(new ConsumerDealThread(record));
        }
      } catch (Exception e) {
        e.printStackTrace();
        consumer.wakeup();
        executor.shutdown();
        try {
          if (!executor.awaitTermination(5000, TimeUnit.MILLISECONDS)) {
            System.out.println("超时，未关闭线程池");
          }
        } catch (InterruptedException e2) {
          e.printStackTrace();
        }
      }
      BlockingQueue<Runnable> queue = ((ThreadPoolExecutor) executor).getQueue();
      System.out.println("队列数量："+queue.size());
    }
  }
}

@Test
public void consumer(){
  try {
    //注释1
    while (true) {
      //注释2
      ConsumerRecords<String, String> records = consumer.poll(100);
      //注释3
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
      }
      try {
        consumer.commitSync();  //注释5
      } catch (CommitFailedException e) {
        log.error("commit failed", e)  //注释6
      }
    }
  } catch (Exception e) {
    log.error("消费者处理数据失败", e);
  } finally {
    //注解4
    consumer.close();
  }
}

@Test
public void commitAsync(){
  try {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
      }
      //注释1
      // consumer.commitAsync();

      //注释2
      consumer.commitAsync((offsets, exception) ->{
        if (exception != null){
          log.error("Commit failed for offsets {}", offsets, exception);
        }
      });
    }

  } catch (Exception e) {
    log.error("消费者处理数据失败", e);
  } finally {
    consumer.close();
  }
}

@Test
public void commitSyncAndAsync(){
  try {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
      }
      //注释1
      consumer.commitAsync();
    }
  } catch (Exception e) {
    log.error("消费者处理数据失败", e);
  } finally {
    try {
      //注释2
      consumer.commitSync();
    }finally {
      consumer.close();
    }
  }
}

@Test
public void customOffsetsCommit() {
  Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
  int count = 1;
  try {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(100);
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
        // 注释1
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1, "no metadata");
        // 注释2
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), offsetAndMetadata);
        // 注释3
        if (count++ % 1000 == 0) {
          // 注释4
          consumer.commitAsync(currentOffsets, null);
        }
      }
    }
  } catch (Exception e) {
    log.error("消费者处理数据失败", e);
  } finally {
    consumer.close();
  }
}

//消费者添加再均衡监听器
consumer.subscribe(Collections.singletonList(TOPIC), new RebalanceListener(consumer));

@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
  System.out.println("分区再均衡，提交当前偏移量：" + currentOffsets);
  consumer.commitSync(currentOffsets);
}

@Service
@Slf4j
public class RebalanceListenerConsumer {

  private static final String TOPIC = "HLJ_TOPIC_JAVA";
  private static final String BROKER_LIST = "localhost:9092";
  private static final String GROUP_ID = "HTEST_GROUP";
  //记录当前偏移量
  private static Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
  private static KafkaConsumer<String, String> consumer = null;

  static {
    Properties properties = initConfig();
    consumer = new KafkaConsumer<>(properties);
    //消费者添加再均衡监听器
    consumer.subscribe(Collections.singletonList(TOPIC), new RebalanceListener(consumer));
  }

  private static Properties initConfig() {
    Properties properties = new Properties();
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
    properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    return properties;
  }

  /**
   * 再均衡监听器
   */
  @Slf4j
  public static class RebalanceListener implements ConsumerRebalanceListener {


    private KafkaConsumer<String, String> consumer;

    /**
     * 初始化方法，传入consumer对象，否则无法调用外部的consumer对象，必须传入
     */
    public RebalanceListener(KafkaConsumer<String, String> consumer) {
      this.consumer = consumer;
    }


    /**
     * 该方法会在再均衡之后和消费者读取之前被调用
     * 使用：在获得新分区后开始读取消息，不需要做其他事情。（最好是消费的时候保存偏移量到数据库中，然后从数据库中获取偏移量的位置）
     */
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
      long committedOffset = -1;
      for (TopicPartition topicPartition : partitions) {
        // 获取该分区已经提交的偏移量
        committedOffset = consumer.committed(topicPartition).offset();
        System.out.println("重新分配分区，提交的偏移量：" + committedOffset);
        // 重置偏移量到上一次提交的偏移量的下一个位置处开始消费
        consumer.seek(topicPartition, committedOffset + 1);
        
        //从数据库中获取偏移量，前提是要在每次消费之后保存到数据库中
        //consumer.seek(partition, getOffsetFromDB(partition));
      }
    }


    /**
      * 该方法会在再均衡开始之前和消费者停止读取之后被调用。如果在这个方法中提交偏移量，则下一个消费者就可以获得读取的偏移量。
      * 使用：在失去分区所有权之前通过 onPartitionsRevoked() 方法来提交偏移量
      * 解释：如果发生再均衡，我们要在即将失去分区所有权时提交偏移量。
      * 要注意，提交的是最近处理过的偏移量，而不是批次中还在处理的最后一个偏移量。因为分区有可能在我们还在处理消息的时候被撤回。
      *        我们要提交所有分区的偏移量，而不只是那些即将失去所有权的分区的偏移量——因为提交的偏移量是已经处理过的，所以不会有什么问题。调用 commitSync() 方法，确保在再均衡发生之前提交偏移量。
         */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
      System.out.println("分区再均衡，提交当前偏移量：" + currentOffsets);
      consumer.commitSync(currentOffsets);
    }

  }


  @Test
  public void consumer() {
    try {
      int count = 1;
      while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
         log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
          OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1, "no metadata");
          currentOffsets.put(new TopicPartition(record.topic(), record.partition()), offsetAndMetadata);
          if (count++ % 1000 == 0) {
            // 注释4
            consumer.commitAsync(currentOffsets, null);
          }
        }
      }
    } catch (Exception e) {
      log.error("消费者处理数据失败", e);
    } finally {
      try {
        //注释2
        consumer.commitSync();
      } finally {
        consumer.close();
      }
    }
  }

}

public void seekToBeginning(Collection<TopicPartition> partitions)  
  
  
//使用  
consumer.seekToBeginning(partitions);

public void seekToEnd(Collection<TopicPartition> partitions)   
  
//使用  
consumer.seekToEnd(partitions);

@Service
@Slf4j
public class SaveOffsetsOnRebalanceConsumer {

    private static final String TOPIC = "HLJ_TOPIC_JAVA";
    private static final String BROKER_LIST = "localhost:9092";
    private static final String GROUP_ID = "HTEST_GROUP";
    //记录当前偏移量
    private static Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
    private static KafkaConsumer<String, String> consumer = null;

    static {
        Properties properties = initConfig();
        consumer = new KafkaConsumer<>(properties);
        //消费者添加再均衡监听器
        consumer.subscribe(Collections.singletonList(TOPIC), new SaveOffsetsOnRebalance(consumer));
        // 注释1
        consumer.poll(0);
        for (TopicPartition partition: consumer.assignment()){
            //注释2：
            // consumer.seek(partition, getOffsetFromDB(partition));
        }

    }

    private static Properties initConfig() {
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        return properties;
    }

    /**
     * 再均衡监听器
     */
    @Slf4j
    public static class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {


        private KafkaConsumer<String, String> consumer;

        /**
         * 初始化方法，传入consumer对象，否则无法调用外部的consumer对象，必须传入
         */
        public SaveOffsetsOnRebalance(KafkaConsumer<String, String> consumer) {
            this.consumer = consumer;
        }


        /**
         * 该方法会在再均衡之后和消费者读取之前被调用
         * 使用：在获得新分区后开始读取消息，不需要做其他事情。
         */
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            for(TopicPartition partition: partitions){
                //注释3：
                // consumer.seek(partition, getOffsetFromDB(partition));
            }
        }

        /**
         * 该方法会在再均衡开始之前和消费者停止读取之后被调用。如果在这个方法中提交偏移量，则下一个消费者就可以获得读取的偏移量。
         * 使用：在失去分区所有权之前通过 onPartitionsRevoked() 方法来提交偏移量
         * 解释：如果发生再均衡，我们要在即将失去分区所有权时提交偏移量。
         * 要注意，提交的是最近处理过的偏移量，而不是批次中还在处理的最后一个偏移量。因为分区有可能在我们还在处理消息的时候被撤回。
         *        我们要提交所有分区的偏移量，而不只是那些即将失去所有权的分区的偏移量——因为提交的偏移量是已经处理过的，所以不会有什么问题。调用 commitSync() 方法，确保在再均衡发生之前提交偏移量。
         */
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            //注释4：
            // commitDBTransaction();
        }

    }


    @Test
    public void consumer() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", 
                 record.topic(), record.partition(), record.offset(), record.key(), record.value());
                // processRecord(record);
                //保存记录结果
                // storeRecordInDB(record);
                //保存位移
                //storeOffsetInDB(record.topic(), record.partition(), record.offset());
            }
          		//注释5
            //提交数据库事务，保存消费的记录以及位移
            // commitDBTransaction();
        }
    }

}

@Test
public void test() {
  final Thread mainThread = Thread.currentThread();
  //注释1：
  Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    log.info("Starting exit...");
    consumer.wakeup();
    try {
      // 主线程继续执行，以便可以关闭consumer
      mainThread.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }));

  try {
    while (true) {
      ConsumerRecords<String, String> records = consumer.poll(1000);
      for (ConsumerRecord<String, String> record : records) {
        log.info("topic=[{}], partition = [{}], offset = [{}], key = [{}], value =[{}]", record.topic(), record.partition(), record.offset(), record.key(), record.value());
      }
      consumer.commitAsync();
    }
  } catch (WakeupException e) {
    //注释2：
    log.info("消费者退出", e);
  } finally {
    //注释3
    consumer.close();
    log.info("Closed consumer and we are done");
  }
}