RocketMQ重试机制及消息幂代码实例解析

更新时间：2020年02月21日 14:36:16 作者：流氓大队长

这篇文章主要介绍了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<MessageExt> 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<MessageExt> 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<String, Object> 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<MessageExt> 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!");
  }
}

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

SpringBoot整合EasyExcel 3.x的完整示例
EasyExcel 是一个基于 Java 的、快速、简洁、解决大文件内存溢出的 Excel 处理工具，它能让你在不用考虑性能、内存的等因素的情况下，快速完成 Excel 的读、写等功能，这篇文章主要介绍了SpringBoot整合EasyExcel3.x的过程,需要的朋友可以参考下
2023-07-07
Eclipse中自动添加注释（两种）
本文主要介绍了Eclipse中自动添加注释的两种方法。具有很好的参考价值，下面跟着小编一起来看下吧
2017-02-02
JDK1.8源码下载及idea2021导入jdk1.8源码的详细步骤
这篇文章主要介绍了JDK1.8源码下载及idea2021导入jdk1.8源码的详细步骤,在文章开头就给大家分享了JDK1.8源码下载地址和下载步骤，告诉大家idea2021.1.3导入JDK1.8源码步骤，需要的朋友可以参考下
2022-11-11
SpringBoot程序的打包与运行的实现
本文主要介绍了SpringBoot程序的打包与运行的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2022-06-06
关于pytorch相关部分矩阵变换函数的问题分析
这篇文章主要介绍了pytorch相关部分矩阵变换函数,包括tensor维度顺序变换BCHW顺序的调整，矩阵乘法相关函数,矩阵乘,点乘,求取矩阵对角线元素或非对角线元素的问题，本文给大家介绍的非常详细，需要的朋友可以参考下
2022-03-03
java数字和中文算数验证码的实现
这篇文章主要介绍了java数字和中文算数验证码的实现，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-07-07
如何自定义MyBatis拦截器更改表名
自定义MyBatis拦截器可以在方法执行前后插入自己的逻辑，这非常有利于扩展和定制 MyBatis 的功能，本篇文章实现自定义一个拦截器去改变要插入或者查询的数据源 ,需要的朋友可以参考下
2023-10-10
Java定义泛型接口和类的方法实例分析
这篇文章主要介绍了Java定义泛型接口和类的方法,结合实例形式分析了泛型相关概念、原理及泛型接口与类的定义实现方法,需要的朋友可以参考下
2019-08-08
Java 深入理解创建型设计模式之原型模式
原型（Prototype）模式的定义如下：用一个已经创建的实例作为原型，通过复制该原型对象来创建一个和原型相同或相似的新对象。在这里，原型实例指定了要创建的对象的种类。用这种方式创建对象非常高效，根本无须知道对象创建的细节
2022-02-02
java实现后台数据显示在前端
这篇文章主要为大家详细介绍了java实现后台数据显示在前端，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2020-02-02

RocketMQ重试机制及消息幂代码实例解析

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具