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

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

在数据消费时,避免数据丢失的方法:如果使用了storm,要开启storm的ackfail机制;如果没有使用storm,确认数据被完成处理之后,再更新offset值。低级API中需要手动控制offset值。

数据重复消费的情况,如果处理?
(1)去重:将消息的唯一标识保存到外部介质中,每次消费处理时判断是否处理过;
(2)不管:大数据场景中,报表系统或者日志信息丢失几条都无所谓,不会影响最终的统计分析结果。
时间: 2024-10-23 00:35:22

Kafka消息保证不丢失和重复消费问题的相关文章

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

Kafka重复消费原因 底层根本原因:已经消费了数据,但是offset没提交. 原因1:强行kill线程,导致消费后的数据,offset没有提交. 原因2:设置offset为自动提交,关闭kafka时,如果在close之前,调用 consumer.unsubscribe() 则有可能部分offset没提交,下次重启会重复消费.例如: try { consumer.unsubscribe(); } catch (Exception e) { } try { consumer.close(); }

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

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

关于MQ的几件小事(三)如何保证消息不重复消费

1.幂等性 幂等(idempotent.idempotence)是一个数学与计算机学概念,常见于抽象代数中. 在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同.幂等函数,或幂等方法,是指可以使用相同参数重复执行,并能获得相同结果的函数.这些函数不会影响系统状态,也不用担心重复执行会对系统造成改变.例如,"setTrue()"函数就是一个幂等函数,无论多次执行,其结果都是一样的.更复杂的操作幂等保证是利用唯一交易号(流水号)实现. 简单来说,幂等性就是一个数据

如何保证消息不被重复消费?(如何保证消息消费时的幂等性)

首先就是比如rabbitmq.rocketmq.kafka,都有可能会出现消费重复消费的问题,正常.因为这问题通常不是mq自己保证的,是给你保证的.然后我们挑一个kafka来举个例子,说说怎么重复消费吧. kafka实际上有个offset的概念,就是每个消息写进去,都有一个offset,代表他的序号,然后consumer消费了数据之后,每隔一段时间,会把自己消费过的消息的offset提交一下,代表我已经消费过了,下次我要是重启啥的,你就让我继续从上次消费到的offset来继续消费吧. 但是凡事总

[转帖]kafka 如何保证数据不丢失

https://www.cnblogs.com/MrRightZhao/p/11498952.html 一般我们在用到这种消息中件的时候,肯定会考虑要怎样才能保证数据不丢失,在面试中也会问到相关的问题.但凡遇到这种问题,是指3个方面的数据不丢失,即:producer consumer 端数据不丢失  broker端数据不丢失下面我们分别从这三个方面来学习,kafka是如何保证数据不丢失的 一.producer 生产端是如何保证数据不丢失的 1.ack的配置策略 acks = 0 生产者发送消息之

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

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

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

程序重启RocketMQ消息重复消费

最近在调试RocketMQ消息发送与消费的Demo时,发现一个问题:只要重启程序,RocketMQ消息就会重复消费. 那么这是什么原因导致的,又该如何解决呢? 经过一番排查,发现程序使用的RocketMQ客户端版本是3.6.2,而测试环境安装的RocketMQ环境的版本是4.1.0.原来是客户端和服务器端版本不一样导致的,消息并没有最终被消费,即没有ACK消息确认,只要程序重启就会重复消费. 解决方案:RocketMQ客户端版本使用与服务器端的同一版本,即4.1.0版本. 划重点:使用Rocke

Kafka消息模型

一.消息传递模型 传统的消息队列最少提供两种消息模型,一种P2P,一种PUB/SUB,而Kafka并没有这么做,巧妙的,它提供了一个消费者组的概念,一个消息可以被多个消费者组消费,但是只能被一个消费者组里的一个消费者消费,这样当只有一个消费者组时就等同与P2P模型,当存在多个消费者组时就是PUB/SUB模型. Kafka 的 consumer 是以pull的形式获取消息数据的. pruducer push消息到kafka cluster ,consumer从集群中pull消息,如下图.该博客主要