【Kafka】——顺序消费、重复消费、消息丢失

逃离我推掉我的手 2022-10-05 11:00 501阅读 1赞

前言

Kafka可以说是为分布式而生的一个消息中间件,功能很强大,提到这个,我们可能就会想到消息中间件常提到的几个问题,消费的顺序性、重复消费、消息丢失等问题,接下来我们一一来看。

一、消费的顺序性

现实场景

  • 数据库中的binlog
  • 一些业务需要,比如希望把某个订单的数据消费是有顺序的

问题描述

生产者在写的时候,其实可以指定一个 key,比如说我们指定了某个订单 id 作为 key,那么这个订单相关的数据,一定会被分发到同一个 partition 中去,写入同一个partion中的数据是一定有顺序的,如果是单线程是没有问题的,但是吞吐量太低了,但是如果是多线程是话,顺序就可能会乱掉。

20172170fab2944ea49f805d37f44648.png

解决办法

  1. 一个 topic,一个 partition,一个 consumer,内部单线程消费,单线程吞吐量太低,一般不会用这个。
  2. 写 N 个内存 queue,具有相同 key 的数据都到同一个内存 queue;然后对于 N 个线程,每个线程分别消费一个内存 queue 即可,这样就能保证顺序性。

4c666ae1f31388da3894d8329bb89525.png

二、重复消费

  1. 消费方幂等操作,重复消费不会产生问题
  2. 对每个partitionID,产生一个uniqueID,.只有这个partition的数据被完全消费,才算成功,否则失败回滚。下次若重复执行,就skip

三、消息丢失

1.生产者数据不丢失

同步模式:配置=1(只有Leader收到,-1所有副本成功,0不等待)。leader partition挂了,数据就会丢失。

解决:设置为-1保证produce写入所有副本算成功
producer.type=sync
request.required.acks=-1

异步模式:当缓冲区满了,如果配置为0(没有收到确认,一满就丢弃),数据立刻丢弃

解决:不限制阻塞超时时间。就是一满生产者就阻塞
producer.type=async
request.required.acks=1
queue.buffering.max.ms=5000
queue.buffering.max.messages=10000
queue.enqueue.timeout.ms = -1
batch.num.messages=200

2.消费者数据不丢失 :流计算,基本数据源不适用。高级数据源以kafka为例,由2种方式:receiver(开启WAL,失败可恢复)和director(checkpoint保证)

3.若是storm在消费,开启storm的ackfail机制;若不是storm,数据处理完更新offset,低级API手动控制offset
4.Kafka发送数据过快,导致服务器网卡流量暴增。或磁盘过忙,出现丢包。
(1) 首先,对kafka进行限速,
(2) 其次启用重试机制,使重试间隔变长。
(3) Kafka设置ack=all,即需要处于ISR(副本列表)的分区都确认,才算发送成功。
props.put(“compression.type”, “gzip”);
props.put(“linger.ms”, “50”);
props.put(“acks”, “all”)表示至少成功发送一次;
props.put(“retries “, 30);
props.put(“reconnect.backoff.ms “, 20000);
props.put(“retry.backoff.ms”, 20000)
5.消费者速度很慢,导致一个session周期(0.1版本是默认30s)内未完成消费。导致心跳机制检测报告出问题。比如消费了的数据未及时提交offset,配置有可能是自动提交
问题场景:1.offset为自动提交,正在消费数据,kill消费者线程,下次重复消费。2.设置自动提交,关闭kafka,close之前,调用consumer.unsubscribed()则由可能部分offset没有提交。3.消费程序和业务逻辑在一个线程,导致offset提交超时。

发表评论

表情:
评论列表 (有 0 条评论,501人围观)

还没有评论,来说两句吧...

相关阅读

    相关 06 MQ重复消费 顺序消费

    重复发送消息,怎么保证消息不被重复消费呢? 让消费消息的操作具有幂等性,一个操作具有幂等性是指这个操作任意多次执行所产生的影响与一次执行的影响相同。具体的实现有: 1.