当前位置:首页 > 服务端 > RabbitMQ重试机制

RabbitMQ重试机制

2022年08月05日 15:15:06服务端2

1、RabbitMQ重试机制的简介

RabbitMQ 不会为未确认的消息设置过期时间,它判断此消息是否需要重新投递给消费者的唯一依据是消费该消息连接是否已经断开,这个设置的原因是 RabbitMQ 允许消费者消费一条消息的时间可以很久很久。

RabbitMQ 的 Web 管理平台上可以看到当前队列中的 “Ready” 状态和 “Unacknowledged” 状态的消息数,分别对应等待投递给消费者的消息数和已经投递给消费者但是未收到确认信号的消息数。如下图:

RabbitMQ重试机制 _ JavaClub全栈架构师技术笔记

注意事项:

如果在处理消息的过程中,消费者的服务器在处理消息的时候出现异常,那么可能这条正在处理的消息就没有完成消息消费,数据就会丢失。为了确保数据不会丢失,RabbitMQ 支持消息确认-ACK。

如果忘记了消息确认,那么后果很严重。当 Consumer 退出时候,Message 会一直重新分发。然后 RabbitMQ 会占用越来越多的内容,由于 RabbitMQ 会长时间运行,因此这个"内存泄漏"是致命的。

RabbitMQ 重试机制核心配置:

spring:
  # 项目名称
  application:
    name: rabbitmq-consumer
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # 重试机制
        retry:
          enabled: true #是否开启消费者重试
          max-attempts: 5 #最大重试次数
          initial-interval: 5000ms #重试间隔时间(单位毫秒)
          max-interval: 1200000ms #重试最大时间间隔(单位毫秒)
          multiplier: 2 #间隔时间乘子,间隔时间*乘子=下一次的间隔时间,最大不能超过设置的最大间隔时间

 

2、RabbitMQ重试机制的实现

下面将通过示例来讲解 RabbitMQ 重试机制的实现。首先需要创建两个 SpringBoot 项目并整合 RabbitMQ 客户端。

2.1 实现消息发送端

(1)创建第一个 SpringBoot 项目( rabbitmq-provider 消息发送项目)。

在pom.xml配置信息文件中,添加相关依赖文件:

<!-- AMQP客户端 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.4.1</version>
</dependency>

在 application.yml 配置文件中配置 RabbitMQ 服务:

spring:
  # 项目名称
  application:
    name: rabbitmq-provider
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

(2)配置队列

在 rabbitmq-provider(消息发送项目)中,配置队列名称,并将队列交由 IoC 管理,代码如下:

package com.pjb.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMQ配置类
 * @author pan_junbiao
 **/
@Configuration
public class RabbitMqConfig
{
    public static final String QUEUE_NAME = "queue_name"; //队列名称
    public static final String EXCHANGE_NAME = "exchange_name"; //交换器名称
    public static final String ROUTING_KEY = "routing_key"; //路由键

    /**
     * 队列
     */
    @Bean
    public Queue queue()
    {
        /**
         * 创建队列,参数说明:
         * String name:队列名称。
         * boolean durable:设置是否持久化,默认是 false。durable 设置为 true 表示持久化,反之是非持久化。
         * 持久化的队列会存盘,在服务器重启的时候不会丢失相关信息。
         * boolean exclusive:设置是否排他,默认也是 false。为 true 则设置队列为排他。
         * boolean autoDelete:设置是否自动删除,为 true 则设置队列为自动删除,
         * 当没有生产者或者消费者使用此队列,该队列会自动删除。
         * Map<String, Object> arguments:设置队列的其他一些参数。
         */
        return new Queue(QUEUE_NAME, true, false, false, null);
    }

    /**
     * Direct交换器
     */
    @Bean
    public DirectExchange exchange()
    {
        /**
         * 创建交换器,参数说明:
         * String name:交换器名称
         * boolean durable:设置是否持久化,默认是 false。durable 设置为 true 表示持久化,反之是非持久化。
         * 持久化可以将交换器存盘,在服务器重启的时候不会丢失相关信息。
         * boolean autoDelete:设置是否自动删除,为 true 则设置队列为自动删除,
         */
        return new DirectExchange(EXCHANGE_NAME, true, false);
    }

    /**
     * 绑定
     */
    @Bean
    Binding binding(DirectExchange exchange, Queue queue)
    {
        //将队列和交换机绑定, 并设置用于匹配键:routingKey
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

(3)创建发送者

在 rabbitmq-provider(消息发送项目)中,创建发送者,利用 rabbitTemplate.convertAndSend() 方法发送消息,代码如下:

package com.pjb;

import com.pjb.config.RabbitMqConfig;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * RabbitMq测试类
 * @author pan_junbiao
 **/
@SpringBootTest
public class RabbitMqTest
{
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    public void sendMessage()
    {
        String message = "您好,欢迎访问 pan_junbiao的博客";
        rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_NAME, RabbitMqConfig.ROUTING_KEY, message);
        System.out.println("消息发送成功!");
    }
}

2.2 实现消息接收端

(1)创建第二个 SpringBoot 项目( rabbitmq-consumer 消息接收项目)。

在pom.xml配置信息文件中,添加相关依赖文件:

<!-- AMQP客户端 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.4.1</version>
</dependency>

在 application.yml 配置文件中配置 RabbitMQ 服务,这里需要配置 RabbitMQ 重试机制:

spring:
  # 项目名称
  application:
    name: rabbitmq-consumer
  # RabbitMQ服务配置
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        # 重试机制
        retry:
          enabled: true #是否开启消费者重试
          max-attempts: 5 #最大重试次数
          initial-interval: 5000ms #重试间隔时间(单位毫秒)
          max-interval: 1200000ms #重试最大时间间隔(单位毫秒)
          multiplier: 2 #间隔时间乘子,间隔时间*乘子=下一次的间隔时间,最大不能超过设置的最大间隔时间

(2)创建接收者

在 rabbitmq-consumer(消息接收项目)中,创建创建接收者,注意,发送者和接收者的 Queue 名称必须一致,否则不能接收消息。

接收者接收到消息后,打印输出消息,然后程序抛出运行时异常,观察现象。代码如下:

package com.pjb.receiver;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * 接收者
 * @author pan_junbiao
 **/
@Component
@RabbitListener(queues="queue_name")
public class Receiver
{
    @RabbitHandler
    public void process(String message)
    {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("接收消息: " + message + " 接收时间:" + sdf.format(new Date()));
        throw new RuntimeException();
    }
}

特别注意:

如果在消息接收端的 application.yml 配置文件中没有添加 RabbitMQ 重试机制的相关配置,当接收端收到消息后程序抛出异常,那么发送端将得不到消息确认(ACK),此时发送端将会循环的发送消息,最终导致内存溢出。

执行结果:

RabbitMQ重试机制 _ JavaClub全栈架构师技术笔记

从上述执行结果来看,当接收端重试5次后,将消息确认(ACK)。

 

作者:pan_junbiao
来源链接:https://blog.csdn.net/pan_junbiao/article/details/113038421

版权声明:
1、JavaClub(https://www.javaclub.cn)以学习交流为目的,由作者投稿、网友推荐和小编整理收藏优秀的IT技术及相关内容,包括但不限于文字、图片、音频、视频、软件、程序等,其均来自互联网,本站不享有版权,版权归原作者所有。

2、本站提供的内容仅用于个人学习、研究或欣赏,以及其他非商业性或非盈利性用途,但同时应遵守著作权法及其他相关法律的规定,不得侵犯相关权利人及本网站的合法权利。
3、本网站内容原作者如不愿意在本网站刊登内容,请及时通知本站(javaclubcn@163.com),我们将第一时间核实后及时予以删除。


本文链接:https://www.javaclub.cn/server/17758.html

标签: RabbitMQ
分享给朋友:

“RabbitMQ重试机制” 的相关文章

中间件RabbitMQ

中间件RabbitMQ

一、前言 RabbitMQ在Windows上安装非常简单,就跟日常安装应用聊天软件一样,因此这里选择Linux环境安装RabbitMQ,并且借助Docker容器进行安装,在使用Docker安装RabbitMQ之前,我们先简单的了解一下几个概念:什么...

MQ--01为什么要选择RabbitMQ ,RabbitMQ简介,各种MQ选型对比

MQ--01为什么要选择RabbitMQ ,RabbitMQ简介,各种MQ选型对比

MQ 是什么?队列是什么,MQ 我们可以理解为消息队列,队列我们可以理解为管道。以管道的方式做消息传递。 场景:     1.其实我们在双11的时候,当我们凌晨大量的秒杀和抢购商品,然后去结算的时候,就会发现,界面会提...

RabbitMQ配置SSL

RabbitMQ配置SSL

引言 主要介绍了如何通过Docker安装RabbitMQ并配置SSL;配置成功之后给出了SpringBoot和Python集成的例子。 生成证书 $ git clone https://github.com/Berico-Technol...

RabbitMQ基本特性

RabbitMQ基本特性

1.互联网大厂为什么选择RabbitMQ? 2.RabbitMQ的高性能之道是如何做到的? 3.什么是AMQP高级协议? 4.AMQP核心概念是什么? 5.RabbitMQ整体架构模型是什么样子? 6.RabbitMQ消息是如何流...

消息中间件RabbitMQ学习笔记---RabbitMQ简介

消息中间件RabbitMQ学习笔记---RabbitMQ简介

消息中间件RabbitMQ学习笔记—RabbitMQ简介 1.RabbitMQ简介 RabbitMQ,俗称“兔子MQ”(可见其轻巧,敏捷),是目前非常热门的一款开源消息中间件,不管是互联网行业还是传统行业都广泛使用(最早是为了解决电信...

rabbitmq权限配置

关于rabbitmq权限的几点: rabbitmq的权限控制通过两层来实现,一是vhost的权限,二是确认有权限访问vhost后,对vhost内资源的权限控制(配置,读,写)   1.默认...

rabbitmq 启动异常

问题点说明:今天早上,开发人员反映Rabbitmq有问题,我试着登录页面访问http://IP:15672,出现了报错信息,因为当时忙着处理,报错页面没有截下来。 后面想重启一下Rabbitmq看能不能解决问题,发现居然重启不了,然后查看了下日志信息,找到了Ra...

linux查看rabbitmq运行状态,RabbitMQ常用命令

RabbitMQ常用命令 rabbitmqctl命令 http://www.rabbitmq.com/man/rabbitmqctl.1.man.html# 1). 服务器启动与关闭 启动: rabbitmq-server –detach...

RabbitMQ工作方式

RabbitMQ工作方式: 工作队列模式 发布订阅模式 路由模式 通配符模式 header转发模式 RPC远程过程调用模式   作者:wondream322 来源链接:https://blog.csdn.net/wang386476890/art...

Linux系统Centos7安装RabbitMQ使用压缩包配置环境变量RabbitMQ 3.8.16.tar.xz Erlang 24

Linux系统Centos7安装RabbitMQ使用压缩包配置环境变量RabbitMQ 3.8.16.tar.xz Erlang 24

文件的下载1、创建两个安装信息的存放文件夹sudo mkdir /usr/local/rabbitmqsudo mkdir /usr/local/erlang  2、下载rabbitmq和erlangcd /usr/local/rab...

发表评论

访客

◎欢迎参与讨论,请在这里发表您的看法和观点。