kafka最初是被LinkedIn设计用来处理log的分布式消息系统,因此它的着眼点不在数据的安全性(log偶尔丢几条无所谓),换句话说kafka并不能完全保证数据不丢失。
尽管kafka官网声称能够保证at-least-once,但如果consumer进程数小于partition_num,这个结论不一定成立。
考虑这样一个case,partiton_num=2,启动一个consumer进程订阅这个topic,对应的,stream_num设为2,也就是说启两个线程并行处理message。
如果auto.commit.enable=true,当consumer fetch了一些数据但还没有完全处理掉的时候,刚好到commit interval出发了提交offset操作,接着consumer crash掉了。这时已经fetch的数据还没有处理完成但已经被commit掉,因此没有机会再次被处理,数据丢失。
如果auto.commit.enable=false,假设consumer的两个fetcher各自拿了一条数据,并且由两个线程同时处理,这时线程t1处理完partition1的数据,手动提交offset,这里需要着重说明的是,当手动执行commit的时候,实际上是对这个consumer进程所占有的所有partition进行commit,kafka暂时还没有提供更细粒度的commit方式,也就是说,即使t2没有处理完partition2的数据,offset也被t1提交掉了。如果这时consumer crash掉,t2正在处理的这条数据就丢失了。
如果希望能够严格的不丢数据,解决办法有两个:
- 手动commit offset,并针对partition_num启同样数目的consumer进程,这样就能保证一个consumer进程占有一个partition,commit offset的时候不会影响别的partition的offset。但这个方法比较局限,因为partition和consumer进程的数目必须严格对应。
- 另一个方法同样需要手动commit offset,另外在consumer端再将所有fetch到的数据缓存到queue里,当把queue里所有的数据处理完之后,再批量提交offset,这样就能保证只有处理完的数据才被commit。当然这只是基本思路,实际上操作起来不是这么简单,具体做法以后我再另开一篇。
相关推荐
Go-consumergroup:采用golang编写的kafka consumer库
这是使用java操作kafka consumer api的一个demo,欢迎下载交流,博客地址:https://blog.csdn.net/qq_26803795
研究了一段时间后,根据网上的例子,做大量的削减及根据需要做出的最简化使用实例,并且加入了获取kafka的server端的状态信息,根据状态信息配置启动时读写位置
KafkaConsumerDemo.java
kafka读取写入数据
向kafka插入数据测试
kafka_hdfs_consumer实现
1、Kafka如何防止数据丢失 1)消费端弄丢数据 消费者在消费完消息之后需要执行消费位移的提交,该消费位移表示下一条需要拉取的消息的位置。Kafka默认位移提交方式是自动提交,但它不是在你每消费一次数据之后就...
Storm集成Kafka 一、整合说明 二、写入数据到Kafka 三、从Kafka中读取数据 从Kafka中读取数据 Storm从Kafka集群读取数据处理
kettle7.1版本整合kafka,kafka插件包含生产者、消费者。直接在kettle安装目录plugins下创建steps目录,并解压下载文件到kettle/plugins/steps目录。具体可查看...
很多时候,客户程序只是希望从Kafka读取数据,不太关心消息offset的处理。同时也希望提供一些语义,例如同一条消息只被某一个Consumer消费(单播)或被所有Consumer消费(广播)。因此,KafkaHightL
使用场景:生产环境海量数据,用kafka-console-consumer 消费kafka某时间段消息用于分析问题,生产环境海量数据,用kafka-console-consumer.sh只能消费全量,文件巨大,无法grep。 代码来源于博主:BillowX_ ,感谢...
sparkStreaming消费数据不丢失,sparkStreaming消费数据不丢失
使用netty实现TCP长链接消息写入kafka以及kafka批量消费数据,数据可以批量进行操作
php-kafka-consumer 主要是对 php_rdkafka 的 consumer 的 API 进行一层封装,增加了原程序中所没有的与 zookeeper 交互的功能。
kettle kafka 消费者插件,在plugins 下新建steps文件夹,把zip文件解压放到里面。
用java实现向kafka写数据以及从kafka消费数据,kafka版本支持0.10以上
.net core 使用kafka推送消息和接收消息,包含生产端和消费端
kafka学习过程,maven工程,包含基础过程、提升过程。可供大家学习一下,里面有详细注释,一个groupid多个Consumer来消费消息和一个Consumer且有多个线程消费