重庆分公司
重庆分公司，新征程启航
为企业提供网站建设、域名注册、服务器等服务
RocketMQ重试机制的示例分析
这篇文章主要介绍RocketMQ重试机制的示例分析，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！
公司主营业务：网站设计制作、做网站、移动网站开发等业务。帮助企业客户真正实现互联网宣传，提高企业的竞争能力。成都创新互联公司是一支青春激扬、勤奋敬业、活力青春激扬、勤奋敬业、活力澎湃、和谐高效的团队。公司秉承以“开放、自由、严谨、自律”为核心的企业文化，感谢他们对我们的高要求，感谢他们从不同领域给我们带来的挑战，让我们激情的团队有机会用头脑与智慧不断的给客户带来惊喜。成都创新互联公司推出马关免费做网站回馈大家。
一.重试机制
1.由于MQ经常处于复杂的分布式系统中，考虑网络波动，服务宕机，程序异常因素，很有可能出现消息发送或者消费失败的问题。因此，消息的重试就是所有MQ中间件必须考虑到的一个关键点。如果没有消息重试，就可能产生消息丢失的问题，可能对系统产生很大的影响。所以，秉承宁可多发消息，也不可丢失消息的原则，大部分MQ都对消息重试提供了很好的支持。
2.RocketMQ为了使用者封装了消息重试的处理流程，无需开发人员手动处理。RocketMQ支持了生产端和消费端两类重试机制。
模拟异常
Consumer端消息消费两种状态：
package com.alibaba.rocketmq.client.consumer.listener;
public enum ConsumeConcurrentlyStatus {
CONSUME_SUCCESS,
RECONSUME_LATER;
private ConsumeConcurrentlyStatus() {
}
}一个是成功（CONSUME_SUCCESS），一个是失败&重试（RECONSUME_LATER）;
Consumer为了保证消息消费成功，只有使用方明确表示消费成功，返回CONSUME_SUCCESS，RocketMQ才会认为消息消费成功。
如果消息消费失败，只要返回ConsumeConcurrentlyStatus.RECONSUME_LATER,RocketMQ就会认为消息消费失败了，需要重新投递。
1.出现异常
package com.wn.consumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
public class MQConsumer {
public static void main(String[] args) throws MQClientException {
//创建消费者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//设置NameServer地址
consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
//设置消费者实例名称
consumer.setInstanceName("consumer");
//订阅topic
consumer.subscribe("wn02","TagA");
//监听消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//获取消息
for (MessageExt msg:list){
System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
}
try {
int i=1/0;
}catch (Exception e){
e.printStackTrace();
//需要重试
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}//消息成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started...");
}
} 2.网络延迟
package com.wn.consumer;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.List;
public class MQConsumer {
public static void main(String[] args) throws MQClientException {
//创建消费者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//设置NameServer地址
consumer.setNamesrvAddr("192.168.138.187:9876;192.168.138.188:9876");
//设置消费者实例名称
consumer.setInstanceName("consumer");
//订阅topic
consumer.subscribe("wn03","TagA");
//监听消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//获取消息
for (MessageExt msg:list){
System.out.println(msg.getMsgId()+"---"+new String(msg.getBody()));
}
//网络延迟
try {
Thread.sleep(600000);
} catch (InterruptedException e) {
e.printStackTrace();
}
//消息成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started...");
}
} 二、消息幂等
1、在什么情况下会发生RocketMQ的消息重复消费
①、当系统的调用链路比较长的时候，比如系统A调用系统B，系统B再把消息发送到RocketMQ中，在系统A调用系统B的时候，如果系统B处理成功，但是迟迟没有将调用成功的结果返回给系统A的时候，系统A就会尝试重新发起请求给系统B，造成系统B重复处理，发起多条消息给RocketMQ造成重复消费；
②、在系统B发送给RocketMQ的时候，也有可能会发生和上面一样的问题，消息发送超时，节骨系统B重试，导致RocketMQ接收到了重读消息；
③、当RocketMQ成功接收到消息，并将消息交给消费者处理，如果消费者消费完成后还没来得及提交offset给RocketMQ，自己宕机或者重启了，那么RocketMQ没有接收到offset，就会认为消费失败了，会重发消息给消费者再次消费；
2、如何解决消息的重复消费
通过幂等性来保证，只要保证重复消息不对结果产生影响，就完美地解决这个问题。
在生产者端保证幂等性，一下两种方式：
①、RocketMQ支持消息查询的功能，只要去RocketMQ查询一下是否已经发送过该条消息就可以了，不存在则发送，存在则不发送；
②、引入redis，在发送消息到RocketMQ成功之后，向Redis中插入一条数据，如果发送重试，则先去Redis查询一个该条消息是否已经发送过了，存在的话就不重复发送消息了；
方法一：RocketMQ消息查询的性能不是特别好，如果在高并发的场景下，每条消息在发送到RocketMQ时都去查询一下，可能会影响接口的性能；
方法二：在一些极端的场景下，Redis也无法保证消息发送成功之后，就一定能写入Redis成功，比如写入消息成功而Redis此时宕机，那么再次查询Redis判断消息是否已经发送过，是无法得到正确结果的；
3、生产者
package com.zn.idempotent;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
/**
* 消息幂等生产者
*/
public class IdempotentProvider {
public static void main(String[] args) throws MQClientException, InterruptedException, RemotingException, MQBrokerException {
//创建一个生产者
DefaultMQProducer producer=new DefaultMQProducer("rmq-group");
//设置NameServer地址
producer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
//设置生产者实例名称
producer.setInstanceName("producer");
//启动生产者
producer.start();
//发送消息
for (int i=1;i<=1;i++){
//模拟网络延迟，每秒发送一次MQ
Thread.sleep(1000);
//创建消息，topic主题名称 tags临时值代表小分类， body代表消息体
Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
//消息的唯一标识
message.setKeys("订单消息："+i);
//发送消息
SendResult sendResult=producer.send(message);
System.out.println("信息幂等问题来了："+sendResult.toString());
}
producer.shutdown();
}
}4、消费者
package com.zn.idempotent;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.LogManager;
import java.util.logging.Logger;
/**
* 消息幂等消费者
*/
public class IdempotentConsumer {
static private Map logMap = new HashMap<>();
public static void main(String[] args) throws MQClientException {
//创建消费者
DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
//设置NameServer地址
consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
//设置实例名称
consumer.setInstanceName("consumer");
//订阅topic
consumer.subscribe("itmayiedu-topic03","TagA");
//监听消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
String key=null;
String msgId=null;
for (MessageExt messageExt:list){
key=messageExt.getKeys();
//判读redis中有没有当前消息key
if (logMap.containsKey(key)) {
// 无需继续重试。
System.out.println("key:"+key+",已经消费，无需重试...");
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
//RocketMQ由于是集群环境，所以产生的消息ID可能会重复
msgId = messageExt.getMsgId();
System.out.println("key:" + key + ",msgid:" + msgId + "---" + new String(messageExt.getBody()));
//将当前key保存在redis中
logMap.put(messageExt.getKeys(),messageExt);
}
try {
int i=5/0;
}catch (Exception e){
e.printStackTrace();
//人工补偿
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//启动消费者
consumer.start();
System.out.println("Consumer Started!");
}
} 以上是“RocketMQ重试机制的示例分析”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注创新互联行业资讯频道！
名称栏目：RocketMQ重试机制的示例分析
转载源于：http://cqcxhl.com/article/pcieie.html