1
消费者pull数据时,出现数据丢失?
- 自动提交offset:可能出现数据丢失,当消费者的数量>partition数量时,出现提交异常。。。多个consumer group 同时消费一个分区的数据,其中一个先提交了,另一个就丢失了。
- 手动提交offset : 防止线程安全问题。
如果希望能够严格的不丢数据,解决办法有两个:
手动commit offset,并针对partition_num启同样数目的consumer进程,这样就能保证一个consumer进程占有一个partition,commit offset的时候不会影响别的partition的offset。但这个方法比较局限,因为partition和consumer进程的数目必须严格对应。
另一个方法同样需要手动commit offset,另外在consumer端再将所有fetch到的数据缓存到queue里,当把queue里所有的数据处理完之后,再批量提交offset,这样就能保证只有处理完的数据才被commit。当然这只是基本思路,实际上操作起来不是这么简单,具体做法以后我再另开一篇。
2
afka作为当下流行的高并发消息中间件,大量用于数据采集,实时处理等场景,我们在享受他的高并发,高可靠时,还是不得不面对可能存在的问题,最常见的就是丢包,重发问题。
2.1 丢包问题
丢包问题:消息推送服务,每天早上,手机上各终端都会给用户推送消息,这时候流量剧增,可能会出现kafka发送数据过快,导致服务器网卡爆满,或者磁盘处于繁忙状态,可能会出现丢包现象。
解决办法:
限速,启用重试机制,重试间隔时间设置长一些,Kafka设置acks=all
首先对kafka进行限速, 其次启用重试机制,重试间隔时间设置长一些,最后Kafka设置acks=all,即需要相应的所有处于ISR的分区都确认收到该消息后,才算发送成功。
检测方法:使用重放机制,查看问题所在。
kafka配置如下:
2.2 重发问题
重发问题:当消费者重新分配partition的时候,可能出现从头开始消费的情况,导致重发问题。当消费者消费的速度很慢的时候,可能在一个session周期内还未完成,导致心跳机制检测报告出问题。
底层根本原因:已经消费了数据,但是offset没提交。
配置问题:设置了offset自动提交
解决办法:至少发一次+去重操作(幂等性)
问题场景:
1.设置offset为自动提交,正在消费数据,kill消费者线程;
2.设置offset为自动提交,关闭kafka时,如果在close之前,调用 consumer.unsubscribe() 则有可能部分offset没提交,下次重启会重复消费;
3.消费kafka与业务逻辑在一个线程中处理,可能出现消费程序业务处理逻辑阻塞超时,导致一个周期内,offset还未提交;继而重复消费,但是业务逻辑可能采用发送kafka或者其他无法回滚的方式;
重复消费最常见的原因:
re-balance问题,通常会遇到消费的数据,处理很耗时,导致超过了Kafka的session timeout时间(0.10.x版本默认是30秒),那么就会re-balance重平衡,此时有一定几率offset没提交,会导致重平衡后重复消费。
去重问题:消息可以使用唯一id标识
保证不丢失消息:
生产者(ack=all 代表至少成功发送一次)
消费者 (offset手动提交,业务逻辑成功处理后,提交offset)
保证不重复消费:
落表(主键或者唯一索引的方式,避免重复数据)
业务逻辑处理(选择唯一主键存储到Redis或者mongdb中,先查询是否存在,若存在则不处理;若不存在,先插入Redis或Mongdb,再进行业务逻辑处理)
网友评论