Kafka重复消费和丢失数据研究

Kafka重复消费原因

底层根本原因:已经消费了数据,但是offset没提交。

原因1:强行kill线程,导致消费后的数据,offset没有提交。

原因2:设置offset为自动提交,关闭kafka时,如果在close之前,调用 consumer.unsubscribe() 则有可能部分offset没提交,下次重启会重复消费。例如:

try {

consumer.unsubscribe();

} catch (Exception e) {

}

try {

consumer.close();

} catch (Exception e) {

}

上面代码会导致部分offset没提交,下次启动时会重复消费。

Kafka Consumer丢失数据原因

猜测:设置offset为自动定时提交,当offset被自动定时提交时,数据还在内存中未处理,此时刚好把线程kill掉,那么offset已经提交,但是数据未处理,导致这部分内存中的数据丢失。

记录offset和恢复offset的方案

理论上记录offset,下一个group consumer可以接着记录的offset位置继续消费。

offset记录方案:

每次消费时更新每个topic+partition位置的offset在内存中,

Map<key, value>,key=topic+‘-‘+partition,value=offset

当调用关闭consumer线程时,把上面Map的offset数据记录到 文件中*(分布式集群可能要记录到redis中)。

下一次启动consumer,需要读取上一次的offset信息,方法是 以当前的topic+partition为key,从上次的Map中去寻找offset。

然后使用consumer.seek()方法指定到上次的offset位置。

说明:

1、该方案针对单台服务器比较简单,直接把offset记录到本地文件中即可,但是对于多台服务器集群,offset也要记录到同一个地方,并且需要做去重处理。

如果线上程序是由多台服务器组成的集群,是否可以用一台服务器来支撑?应该可以,只是消费慢一点,没多大影响。

2、如何保证接着offset消费的数据正确性

为了确保consumer消费的数据一定是接着上一次consumer消费的数据,

consumer消费时,记录第一次取出的数据,将其offset和上次consumer最后消费的offset进行对比,如果相同则继续消费。如果不同,则停止消费,检查原因。

时间: 2024-10-14 02:15:40

Kafka重复消费和丢失数据研究的相关文章

【troubleshooting】记一次Kafka集群重启导致消息重复消费问题处理记录

因需要重启了Kafka集群,重启后发现部分topic出现大量消息积压,检查consumer日志,发现消费的数据竟然是几天前的.由于平时topic消息基本上无积压,consumer消费的数据都是最新的,明显是consumer在重新消费之前已经消费过的数据. 处理方法:将Kafka topic中consumer已经消费的offset值设置为最大值步骤如下:1.从Kafka查询出目前堵塞的topic消息队列中,最大的offset值(其实从Kafka的管理页面上也可以看到这值):命令:./kafka-r

Kafka在高并发的情况下,如何避免消息丢失和消息重复?kafka消费怎么保证数据消费一次?数据的一致性和统一性?数据的完整性?

1.kafka在高并发的情况下,如何避免消息丢失和消息重复? 消息丢失解决方案: 首先对kafka进行限速, 其次启用重试机制,重试间隔时间设置长一些,最后Kafka设置acks=all,即需要相应的所有处于ISR的分区都确认收到该消息后,才算发送成功 消息重复解决方案: 消息可以使用唯一id标识 生产者(ack=all 代表至少成功发送一次) 消费者 (offset手动提交,业务逻辑成功处理后,提交offset) 落表(主键或者唯一索引的方式,避免重复数据) 业务逻辑处理(选择唯一主键存储到R

Kafka消息保证不丢失和重复消费问题

使用同步模式的时候,有3种状态保证消息被安全生产,在配置为1(只保证写入leader成功)的话,如果刚好leader partition挂了,数据就会丢失.还有一种情况可能会丢失消息,就是使用异步模式的时候,当缓冲区满了,如果配置为0(还没有收到确认的情况下,缓冲池一满,就清空缓冲池里的消息),数据就会被立即丢弃掉. 在数据生产时避免数据丢失的方法: 只要能避免上述两种情况,那么就可以保证消息不会被丢失.就是说在同步模式的时候,确认机制设置为-1,也就是让消息写入leader和所有的副本.还有,

相同数据源情况下,使用Kafka实时消费数据 vs 离线环境下全部落表后处理数据,结果存在差异

原因分析: 当某个consumer宕机时,消费位点(例如2s提交一次)尚未提交到zookeeper,此时Kafka集群自动rebalance后另一consumer来接替该宕机consumer继续消费,因为先前宕机consumer最近的消费位点尚未提交,导致数据重复消费 突发流量.跨机房(网络请求延时高).网络不稳定,出现丢包现象 业务逻辑有偏差 常见丢包现象如突然掉线.页面卡住.视频卡住.图片加载卡主等,使用Ping测量丢包的最佳方法是向一个IP地址发送大量的Ping命令,然后检查没有应答的那些

Flume简介与使用(三)——Kafka Sink消费数据之Kafka安装

前面已经介绍了如何利用Thrift Source生产数据,今天介绍如何用Kafka Sink消费数据. 其实之前已经在Flume配置文件里设置了用Kafka Sink消费数据 agent1.sinks.kafkaSink.type = org.apache.flume.sink.kafka.KafkaSink agent1.sinks.kafkaSink.topic = TRAFFIC_LOG agent1.sinks.kafkaSink.brokerList = 10.208.129.3:90

关于kafka重新消费数据问题

我们在使用consumer消费数据时,有些情况下我们需要对已经消费过的数据进行重新消费,这里介绍kafka中两种重新消费数据的方法. 1. 修改offset 我们在使用consumer消费的时候,每个topic会产生一个偏移量,这个偏移量保证我们消费的消息顺序且不重复.Offest是在zookeeper中存储的,我们可以设置consumer实时或定时的注册offset到zookeeper中.我们修改这个offest到我们想重新消费的位置,就可以做到重新消费了.具体修改offest的方法这里就不详

kafka一直rebalance故障,重复消费

今天我司线上kafka消息代理出现错误日志,异常rebalance,而且平均间隔2到3分钟就会rebalance一次,分析日志发现比较严重.错误日志如下 08-09 11:01:11 131 pool-7-thread-3 ERROR [] - commit failed org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already r

kafka查看消费数据

一.如何查看 在老版本中,使用kafka-run-class.sh 脚本进行查看.但是对于最新版本,kafka-run-class.sh 已经不能使用,必须使用另外一个脚本才行,它就是kafka-consumer-groups.sh 普通版 查看所有组 要想查询消费数据,必须要指定组.那么线上运行的kafka有哪些组呢?使用以下命令: bin/kafka-consumer-groups.sh --bootstrap-server kafka-1.default.svc.cluster.local

RocketMQ(消息重发、重复消费、事务、消息模式)

RocketMQ基础:https://github.com/apache/rocketmq/tree/rocketmq-all-4.5.1/docs/cn 分布式消息系统作为实现分布式系统可扩展.可伸缩性的关键组件,需要具有高吞吐量.高可用等特点.而谈到消息系统的设计,就回避不了两个问题: 消息的顺序问题 消息的重复问题 RocketMQ作为阿里开源的一款高性能.高吞吐量的消息中间件,它是怎样来解决这两个问题的?RocketMQ 有哪些关键特性?其实现原理是怎样的? 关键特性以及其实现原理 一.