Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("group.id", "CountryCounter");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);

nsumer.subscribe(Collections.singletonList("customerCountries"));

consumer.subscribe("test.*");

try {
   while (true) {  //1)
       ConsumerRecords<String, String> records = consumer.poll(100);  //2)
       for (ConsumerRecord<String, String> record : records)  //3)
       {
           log.debug("topic = %s, partition = %s, offset = %d,
              customer = %s, country = %s\n",
              record.topic(), record.partition(), record.offset(),
              record.key(), record.value());
           int updatedCount = 1;
           if (custCountryMap.countainsValue(record.value())) {
               updatedCount = custCountryMap.get(record.value()) + 1;
           }
           custCountryMap.put(record.value(), updatedCount)
           JSONObject json = new JSONObject(custCountryMap);
           System.out.println(json.toString(4))
       }
   }
} finally {
      consumer.close(); //4
}

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("topic = %s, partition = %s,
        offset = %d, customer = %s, country = %s\n",
        record.topic(), record.partition(), record.offset(),
        record.key(), record.value());
    }
    consumer.commitAsync(new OffsetCommitCallback() {
        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
            if (e != null)
                log.error("Commit failed for offsets {}", offsets, e);
        } 
    });
}

try {
    while (true) {
       ConsumerRecords<String, String> records = consumer.poll(100);
       for (ConsumerRecord<String, String> record : records) {
           System.out.printf("topic = %s, partition = %s, offset = %d,
           customer = %s, country = %s\n",
           record.topic(), record.partition(),
           record.offset(), record.key(), record.value());
       }
       
       consumer.commitAsync();
    }
} catch (Exception e) {
    log.error("Unexpected error", e);
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
int count = 0;

....

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
    {
        System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());

        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()+1, "no metadata"));
        if (count % 1000 == 0)
            consumer.commitAsync(currentOffsets, null);
        count++;
} }

private Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();

private class HandleRebalance implements ConsumerRebalanceListener {
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    }
    
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Lost partitions in rebalance.
          Committing current
        offsets:" + currentOffsets);
        consumer.commitSync(currentOffsets);
    }
}

try {
    consumer.subscribe(topics, new HandleRebalance());
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records)
        {
             System.out.printf("topic = %s, partition = %s, offset = %d, customer = %s, country = %s\n", record.topic(), record.partition(), record.offset(), record.key(), record.value());
             currentOffsets.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset()+1, "no metadata"));
        }
        consumer.commitAsync(currentOffsets, null);
    }
} catch (WakeupException e) {
    // ignore, we're closing
} catch (Exception e) {
   log.error("Unexpected error", e);
} finally {
   try {
       consumer.commitSync(currentOffsets);
   } finally {
       consumer.close();
       System.out.println("Closed consumer and we are done");
   }
}

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records)
    {
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()), record.offset());
        processRecord(record);
        storeRecordInDB(record);
        consumer.commitAsync(currentOffsets);
    }
}

public class SaveOffsetsOnRebalance implements ConsumerRebalanceListener {
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        //在消费者负责的分区被回收前提交数据库事务，保存消费的记录和位移
        commitDBTransaction();
    }
    
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        //在开始消费前，从数据库中获取分区的位移，并使用seek()来指定开始消费的位移
        for(TopicPartition partition: partitions)
            consumer.seek(partition, getOffsetFromDB(partition));
    } 
}

    consumer.subscribe(topics, new SaveOffsetOnRebalance(consumer));
    //在subscribe()之后poll一次，并从数据库中获取分区的位移，使用seek()来指定开始消费的位移
    consumer.poll(0);
    for (TopicPartition partition: consumer.assignment())
        consumer.seek(partition, getOffsetFromDB(partition));

    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records)
        {
            processRecord(record);
            //保存记录结果
            storeRecordInDB(record);
            //保存位移
            storeOffsetInDB(record.topic(), record.partition(), record.offset());
        }
        //提交数据库事务，保存消费的记录以及位移
        commitDBTransaction();
    }

//注册JVM关闭时的回调钩子，当JVM关闭时调用此钩子。
Runtime.getRuntime().addShutdownHook(new Thread() {
          public void run() {
              System.out.println("Starting exit...");
              //调用消费者的wakeup方法通知主线程退出
              consumer.wakeup();
              try {
                  //等待主线程退出
                  mainThread.join();
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          } 
});

...

try {
    // looping until ctrl-c, the shutdown hook will cleanup on exit
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        System.out.println(System.currentTimeMillis() + "--  waiting for data...");
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s\n",record.offset(), record.key(), record.value());
        }
        for (TopicPartition tp: consumer.assignment())
            System.out.println("Committing offset at position:" + consumer.position(tp));
        consumer.commitSync();
    }
} catch (WakeupException e) {
    // ignore for shutdown
} finally {
    consumer.close();
    System.out.println("Closed consumer and we are done");
}

/**
 * @Description:
 * @author: LinQin
 * @date: 2019/11/15
 */
@Component
@Slf4j
public class KafkaPollTask implements Runnable {

    @Autowired
    private KafkaProperties KafkaProperties;

    /**
     * 这里的消费者应该全局创建好
     * 总消费者数目 = 主题数 * 消费者数目
     * 在有线程池注入该线程
     * z
     */
    private KafkaConsumer kafkaConsumer;
    /**
     * kafka分区是否发生重均衡的标志
     */
    private final AtomicBoolean rebalanceFlag = new AtomicBoolean(false);

    /**
     * 检查是否发生重均衡,如果是,则在跳过此次的拉取,并将标志重置为false
     *
     * @return
     */
    public boolean checkHappenRebance() {
        return rebalanceFlag.compareAndSet(true, false);
    }


    private final Map<TopicPartition, OffsetAndMetadata> offsetInfo = Maps.newConcurrentMap();

    @Override
    public void run() {
        // 初始化消费者
        Map<String, Object> configs = KafkaProperties.buildConsumerProperties();
        kafkaConsumer = new KafkaConsumer<>(configs);
        // 给消费者注册 rebalance 监听
        kafkaConsumer.subscribe(Collections.singleton("topic"), new KafkaRebalanceListener(rebalanceFlag));

        try {
            while (true) {
                ConsumerRecords<byte[], byte[]> records = kafkaConsumer.poll(Duration.ofMillis(60 * 1000));
                records.partitions().forEach(part ->{
                    if (checkHappenRebance()) {
                        // 分区 正在rebalance，中断拉取，防止数据重复消费
                        log.warn("partition rebalance; it will be break,part:{}", part);
                    } else {
                        AtomicInteger count = new AtomicInteger();
                        records.records(part).forEach(record -> {
                            count.getAndIncrement();
                            // 记录当前的偏移量
                            offsetInfo.put(new TopicPartition(record.topic(), record.partition()),
                                    new OffsetAndMetadata(record.offset() + 1, UUID.randomUUID().toString()));
                            //  具体业务处理
                            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                            // 处理成功后提交偏移量,分布式下异步提交失败会重试，导致位移回滚提交。
                            // 解决方案：每次发起异步提交时增加此序号，并且将此时的序号作为参数传给回调方法；当消息提交失败回调时，
                            // 检查参数中的序号值与全局的序号值，如果相等那么可以进行重试提交，否则放弃（因为已经有更新的位移提交了）
//                            正常情况下，偶然的提交失败并不是什么大问题，因为后续的提交成功就可以了，所以这个回调可以为null
                            if (count.get() % 1000 == 0) {
                                // 批次提交
                                kafkaConsumer.commitAsync(offsetInfo, (offsets, e) -> {
                                    if (e != null) {
                                        log.error("Commit failed for offsets {}", offsets, e);
                                    }
                                });
                                count.compareAndSet(1000, 0);
                            }
                            offsetInfo.clear();
                        });
                    }
                });


            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 我们使用异步提交来提高性能，但最后使用同步提交来保证位移提交成功。
            kafkaConsumer.commitSync(offsetInfo);
            kafkaConsumer.close();
        }
    }
}

/**
 * @Description: kafka分区rebalace监听器，该类中设置一个 标识，用来处理rebalace后可能会导致的一些问题，如位移提交失败导致的重复汇
 * @author: LinQin
 * @date: 2019/11/15
 */
@Slf4j
public class KafkaRebalanceListener implements ConsumerRebalanceListener {

    private AtomicBoolean rebalanceFlag;

    public KafkaRebalanceListener(AtomicBoolean rebalanceFlag) {
        this.rebalanceFlag = rebalanceFlag;
    }

    /**
     * 发生rebalance时更改标志，在创建消费者的时候指定使用该监听器
     * @param partitions
     */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        partitions.forEach(partition -> {
            log.info("topic {} - partition {} revoked.", partition.topic(), partition.partition());
            rebalanceFlag.set(true);
        });
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        partitions.forEach(partition -> {
            log.info("topic {} - partition {} assigned.", partition.topic(), partition.partition());
        });
    }
}