Kafka重启Consumer后抓不到数据的问题

使用kafka时，经常会遇到这样的问题：
重启某一消费者程序后，一条数据都抓不到了，但是队列中明明还有很多未被消费的消息，这是怎么回事呢？

先说原因，四个字：重复消费

下面进行详细分析：
“消费确认”是所有消息中间件都要解决的一个问题，在kafka中涉及到两个消费位置：
（1）当前取消息所在的consume offset；
（2）程序处理完毕发送ack（确认字符）后所确定的committed offset。
很显然，在异步模式下，committed offset要落后于consume offset。假如consumer挂了重启，那么它将从commited offset位置处开始重新消费，而不是consume offset位置，这也就意味着很可能重复消费，所以会导致一条数据也抓不到。

如何避免重复消费呢，大家知道，在Kafka客户端中，有三种消费确认策略：

         自动地、周期性地ack
         properties props = new Properties();
         props.put("enable.auto.commit", "true");
         props.put("auto.commit.interval.ms", "1000");

        consumer.commitSync()       //调用commitSync，手动同步ack。每处理完一条消息，commitSync一次

        consumer.commitASync()     //手动异步ack

自动地、周期地ack肯定会出现如上重复消费问题，那么依靠上面的consumer.commitSync()，我们每处理完一条消息，就发送一次commitSync()，这样是不是就可以解决重复消费呢？
答案是否定的。如下代码：

   ConsumerRecords<String, String> records = consumer.poll(1000);
   for (ConsumerRecord<String, String> record : records) {
       buffer.add(record);
   }
   if (buffer.size() > minBatchSize) {
       insertIntoDB(buffer);        //操作一存到db
       consumer.commitSync();       //操作二同步发送ack
       buffer.clear();
   }
}

操作一和操作二做不到原子操作，如果在操作一完成，但进行操作二的时候挂了，服务器再次重启，消息仍然会重复消费。
那么怎么解决这个问题呢？
答案就是自己保存commited offset，而不是依赖kafka的集群保存commited offset，把消息的处理和保存offset做成一个原子操作。
如何将消息的处理和保存offset做成一个原子操作呢，Kafka的官方文档列举了自己保存offset的两种使用场景：

image.png

要自己保存committed offset，就要做到以下几个操作：

Configure enable.auto.commit=false   //禁用自动ack
Use the offset provided with each ConsumerRecord to save your position. //每次取到消息，把对应的offset存下来
On restart restore the position of the consumer using seek(TopicPartition, long).//下次重启，通过consumer.seek函数，定位到自己保存的offset，从那开始消费

Kafka本身的机制只能保证消息不漏，即"at least once"，而通过自己来保存committed offset，我们可以实现消费端的消息不重，即"exactly once"，达到消息不重不丢的目的。