安科网

RocketMQ重试机制和消息幂

meilongwhpu

meilongwhpu

2020-02-18

关注 关注

一.重试机制

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!");
    }
}
 

rocketmq alibaba 科技新闻

meilongwhpu

meilongwhpu

0 关注 0 粉丝 0 动态

关注 关注

相关推荐

Java操作RocketMQ

// producerGroup：一般发送同样消息的Producer，归为同一个Group，应用必须设置，并保证命名唯一。Message msg = new Message("itmayiedu-topic", // topic 主题名

lypgcs 2020-02-18

消息疯狂堆积！RocketMQ 出 Bug 了?

用过 MQ 的同学，可能会遇到过消息堆积的问题。而我最近也踩上了这个坑，但是发现结果竟然是这么一个意料之外的原因而导致的。那一晚月和风高，我正准备踏上回家的路，突然收到告警短信轰炸！于是乎赶回公司赶紧打开电脑，登上 RocketMQ 后台查看

LCFlxfldy 2020-08-17

深入理解RocketMQ延迟消息

延迟消息是实际开发中一个非常有用的功能，本文第一部分从整体上介绍秒级精度延迟消息的实现思路，在第二部分结合RocketMQ的延迟消息实现，进行细致的讲解，点出关键部分的源码。第三步介绍延迟消息与消息重试的关系。Broker判断这是一个延迟消息后，将其通过临

IT农场 2020-11-13

RabbitMQ都写了，RocketMQ怎么能落下？

整体架构最近看到了我在Github上写的rabbitmq-examples陆续被人star了，就想着写个rocketmq-examples。对rabbitmq感兴趣的小伙伴可以看我之前的文章。下面把RocketMQ的各个特性简单介绍一下，这样在用的时候心里

ljcsdn 2020-07-27

(五)消息队列

此时，为了提高用户体验和吞吐量，其实可以异步地调用系统B、C、D的接口。所以，我们可以通过引入消息中间件，再系统A执行完后将数据写入到消息队列中直接返回，而BCD系统则负责监听执行。通过以上分析可知，消息中间件最佳选型为RocketMQ。

LCFlxfldy 2020-07-05

RocketMQ(2)---核心概念、特性、使用等

对于RocketMQ而言，感觉官方提供的东西还是可以的：https://github.com/apache/rocketmq/tree/master/docs/cn

lypgcs 2020-06-27

RocketMq producer 发送一条消息所经过的流程

　　RocketMq producer 在发送一条消息时候，从 producer --nameSrv -- Broker 中间经过了什么样子的数据交互。从NameSrv 中获取topic 配置的相关信息，比如 broker 地址，队列数 之类的。

陈晨软件五千言 2020-06-17

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

qingyuerji 2020-06-14

RocketMQ系列（四）顺序消费

折腾了好长时间才写这篇文章，顺序消费，看上去挺好理解的，就是消费的时候按照队列中的顺序一个一个消费；而并发消费，则是消费者同时从队列中取消息，同时消费，没有先后顺序。RocketMQ也有这两种方式的实现，但是在实践的过程中，就是不能顺序消费，好不容易能够实

MojitoBlogs 2020-06-14

RocketMQ系列（五）广播与延迟消息

今天要给大家介绍RocketMQ中的两个功能，一个是“广播”，这个功能是比较基础的，几乎所有的mq产品都是支持这个功能的；另外一个是“延迟消费”，这个应该算是RocketMQ的特色功能之一了吧。广播是把消息发送给订阅了这个主题的所有消费者。首先，消费者端的

lypgcs 2020-06-14

RocketMQ系列（一）基本概念

RocketMQ是阿里出品的一款开源的消息中间件，让其声名大噪的就是它的事务消息的功能。在企业中，消息中间件选择使用RocketMQ的还是挺多的，这一系列的文章都是针对RocketMQ的，咱们先从RocketMQ的一些基本概念和环境的搭建开始聊起。Rock

陈晨软件五千言 2020-06-14

RocketMq 在Netty 下是如何进行消息封装传输

　　要了解rocketMq 需要知道 数据在 rocketMq 中 是如何进行传输，在底层的结构到底是一个什么亚子，这个需要我们对Netty 对字符编解码有一些了解。　　我们从生产者发送消息，broker 接收消息 为例，来开展底层对消息结构。　　1. 如

meilongwhpu 2020-06-13

RocketMQ

然后执行‘start mqnamesrv.cmd’，启动NAMESERVER。成功后会弹出提示框，此框勿关闭。#set it false if you don‘t want use dashboard.default true. #set the mess

陈晨软件五千言 2020-06-11

RocketMQ集群监控平台搭建

RocketMQ有一个对其扩展的开源项目incubator-rocketmq-externals，这个项目中有一个子模块叫rocketmq-console，这个便是管理控制台项目了，先将incubator-rocketmq-externals拉到本地，因为

qingyuerji 2020-06-09

RocketMQ系列（三）消息的生产与消费

前面的章节，我们已经把RocketMQ的环境搭建起来了，是一个两主两从的异步集群。接下来，我们就看看怎么去使用RocketMQ，在使用之前，先要在NameServer中创建Topic，我们知道RocketMQ是基于Topic的消息队列，在生产者发送消息的时

MojitoBlogs 2020-06-09

RocketMQ事务消息

RocketMQ事务消息是指应用本地事务和发送消息操作可以被定义到全局事务中，要么同时成功，要么同时失败。RocketMQ的事务消息提供类似 X/Open XA 的分布事务功能，通过事务消息能达到分布式事务的最终一致。Producer 已经把消息成功发送到

meilongwhpu 2020-06-08

聊一聊高并发高可用那些事 - Kafka篇

偶尔推出一个优惠活动时，请求量极速上升。看了几个招聘网站，提到较多的消息队列有：RabbitMQ、RocketMQ、Kafka 以及 Redis 的消息队列和发布订阅模式。在高可用环境中，Kafka 需要部署多台，避免 Kafka 宕机后，服务无法访问。K

meilongwhpu 2020-06-08

RocketMQ-Namesrv源码解析

Name Server是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。Broker部署相对复杂，Broker分为Master和Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master和Slave

lypgcs 2020-06-07

深度剖析 Kafka/RocketMQ 顺序消息的一些坑

我觉得这个问题问得很频繁，而且非常经典，在这里我就以 Kafka 为例子，说说我对 Kafka 顺序消息的一些理解吧，如有理解不对的地方麻烦留言指点一下。针对以上第二点，下面盘点一下 Kafka 集群中有哪些意外情况会打乱消息的顺序。

MojitoBlogs 2020-06-04

RocketMQ源码 — 九、 RocketMQ延时消息

上一节消息重试里面提到了重试的消息可以被延时消费，其实除此之外，用户发送的消息也可以指定延时时间，然后在指定延时时间之后投递消息，然后被consumer消费。broker收到消息在准备将消息写入存储的时候，判断是延时消息则更改Message的topic为延

meilongwhpu 2020-05-30
meilongwhpu

meilongwhpu

W3CSchool教程
HTML 教程
CSS 教程
Bootstrap 教程
Javascript 教程
jQuery 教程
后端教程
C 教程
Java 教程
PHP 教程
Python 教程
Go 教程
移动开发
Android 教程
Swift 教程
Kotlin 教程
jQuery Mobile 教程
ionic 教程
关于我们
新闻动态
联系方式
招聘英才
安科实验室
帮助与反馈

安科网(Ancii)，中国第一极客网

Copyright © 2013 - 2019 Ancii.com

京ICP备18063983号 京公网安备11010802014868号