使用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"，达到消息不重不丢的目的。
网友评论