首先说下消息重复发送的原因:

1. 消息异常重复发送
2. 消费消息抛出异常
3. 重平衡
4. 清理长时间消费的消息
5. 消费者提交offset失败
6. 主从同步offset失败
7. 服务端持久化offset失败

在说正面的失败原因前,我们先看下rocketmq前提的基础知识

RocketMQ发送消费消息的基本原理(按照FIFO算法)

1. 生产者发送消息,
2. 通过负载均衡算法,选择发送给一个Broker下面的topic队列QUEUE,
3. 当中首先跟这个队列建立连接,随后把消息发送给此QUEUE
4. 若消费者消费到此queue,消息就被消费啦!~

简单说明下:

• Broker: 服务端
• Topic:一类消息集合的名字
• Queue:就是Topic集合中的成员(队列),存储消息的===> 咋理解呢?就是大头兵,真正干活的,存消息的

上面的基本原理是建立在正常的情况下,但是(哈哈,最怕但是了~),就是我们本文的核心了!异常情况下呢?

一、消息异常重复发送

比如消息发送超时,响应超时,中途中断_等等

这个时候我们的RocketMQ为了保证消息成功发送,就会进行消息发送的重试操作,一般会重试两次

//重试次数默认两次,三次只对同步调用生效
private  int  retryTimesWhenSendFailed=2;

你可能会问?怎么进行重试操作呢?

选择另一台机器的QUEUE来发送就行啦!

这个重试操作:好处就是很好保证消息的发送成功,但是会带来的问题就是消息重复发送

eg:

mq第一次发送消息,

但是当时网速很慢,恰好过了超时时间,到达服务端,服务端接受并处理了,

但是发送端因为超时机制,又再次发送消息,这就导致了消息相同重复问题

(好家伙,有没有像计算机握手画面~哈哈)

二、消费消息抛出异常

我们在使用RocketMQ时,当我们处于并发消费消息的模式下,我们就需要:

1. 实现接口:MessageListenerConcurrencyly(目的:处理消息)

2. 其次:

   • 当消费者get到msg后,

   • 就会调用MessageListenerConcurrencyly的实现,

   • 传入需要消费的消息集合msgs(这玩意是核心)

上面代码,当消息出现异常的时候:

status=null,我们会看到status被设置为RECONSUME_LATER(稍后重新消费)

所以一旦抛出异常,那么消息就可以被重复消费!重复消费!!重复消费!!!(重要的说三遍!!!)

你可能会说?这不是正常的操作嘛!抛出异常,消息都没消费成功,不重复消费咋整??

嗯哼,你提问的这个是对的哈,但是(哈哈,有转折了哈!)

我们往前捋一捋,我们消费时,服务端队列我们传入的是整个集合(关键点就在这!!!)

这意味着什么意思呢?

当然使整个消息集合都重新消费(有没有反应过来??没有?我们继续往下看啊!~)

消息处理之后，不论是成功还是异常，都需要对结果进行处理，代码如下:

链家