1.前言
本次的案例和相关配置是基于spring ,具体的版本如下
spring.5.1.5
Spring-kafka-2.2.4
spring-clients-2.0.1
2.消息丢失
2.1 消息为什么会丢失
2.1.1 消费相关参数说明
enable.auto.commit:表示消费者会周期性自动提交消费的offset。默认值true。
auto.commit.interval.ms:在enable.auto.commit为true的情况下, 自动提交的间隔。默认值5秒。
max.poll.records:单次消费者拉取的最大数据条数,默认值500。
max.poll.interval.ms:表示若在阈值时间之内消费者没有消费完上一次poll的消息,
consumer client会主动向coordinator发起LeaveGroup请求,触发Rebalance(再平衡调整);
然后consumer重新发送JoinGroup请求。
session.timeout.ms:group Coordinator(协调者)检测consumer发生崩溃所需的时间。
在这个时间内如果Coordinator未收到Consumer的任何消息,那Coordinator就认为Consumer挂了。默认值10秒。
heartbeat.interval.ms:标识Consumer给Coordinator发一个心跳包的时间间隔。heartbeat.interval.ms越小,发的心跳包越多。默认值3秒
2.1.2 案例
enable.auto.commit:true 表示消费者会周期性自动提交消费的offset。
如果消费者使用以上的配置,就可能会出现消息丢失,大概流程如下;
2.2 解决办法
手动提交offset.具体的代码如下
2.2.1 消费者配置文件
参数解释:
earliest
当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
latest
当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,消费新产生的该分区下的数据
none
topic各分区都存在已提交的offset时,从offset后开始消费;只要有一个分区不存在已提交的offset,则抛出异常
public enum AckMode {
// 当每一条记录被消费者监听器(ListenerConsumer)处理之后提交
RECORD,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后提交
BATCH,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,距离上次提交时间大于TIME时提交
TIME,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后,被处理record数量大于等于COUNT时提交
COUNT,
// TIME | COUNT 有一个条件满足时提交
COUNT_TIME,
// 当每一批poll()的数据被消费者监听器(ListenerConsumer)处理之后, 手动调用Acknowledgment.acknowledge()后提交
MANUAL,
// 手动调用Acknowledgment.acknowledge()后立即提交
MANUAL_IMMEDIATE,
}
2.2.2 监听器示例代码
3. 消息重复消费
3.1 消息重复消费出现的原因
原因1:
消费者宕机、重启或者被强行kill进程,导致消费者消费的offset没有提交。
原因2:
设置enable.auto.commit为true,如果在关闭消费者进程之前,取消了消费者的订阅,则有可能部分offset没提交,下次重启会重复消费。
原因3:
消费后的数据,当offset还没有提交时,Partition就断开连接。比如,通常会遇到消费的数据,处理很耗时,导致超过了Kafka的session timeout.ms时间,那么就会触发reblance重平衡,此时可能存在消费者offset没提交,会导致重平衡后重复消费。
3.2 消息重复消费解决办法
方法1:
思路是提高消费能力,提高单条消息的处理速度,例如对消息处理中比 较耗时的步骤可通过异步的方式进行处理、利用多线程处理等。在缩短单条消息消费时常的同时,根据实际场景可将max.poll.interval.ms值设置大一点,避免不 必要的rebalance,此外可适当减小max.poll.records的值,默认值是500,可根 据实际消息速率适当调小。这种思路可解决因消费时间过长导致的重复消费问题, 对代码改动较小,但无法绝对避免重复消费问题。
方法2:
思路是引入单独去重机制,例如生成消息时,在消息中加入唯一标识符如消息id等。在消费端,我们可以保存最近的1000条消息id到redis或mysql表中,配置max.poll.records的值小于1000。在消费消息时先通过前置表去重后再进行消息的处理。