使用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的两种使用场景:
要自己保存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",达到消息不重不丢的目的。