RabbitMQ实现消息队列延迟功能

梦里梦外; 2021-07-24 23:39 567阅读 0赞

要实现 RabbitMQ 的消息队列延迟功能，一般采用官方提供的插件 “rabbitmq\_delayed\_message\_exchange” 来实现。但 RabbitMQ 版本必须是 3.5.8 以上才支持该插件，否则得用其“死信”功能。

# 1、安装RabbitMQ延迟插件 #

**（1）检查插件**

使用 rabbitmq-plugins list 命令用于查看RabbitMQ安装的插件。

打开一个新的cmd窗口（按快捷键“Win+R”，输入“cmd”）。

**运行命令：**rabbitmq-plugins list

检查RabbitMQ插件安装情况：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70][]

**（2）下载插件**

如果没有安装插件，则直接访问官网进行下。

> **官方下载地址：**[https://www.rabbitmq.com/community-plugins.html][https_www.rabbitmq.com_community-plugins.html]

![20210116160358308.jpg][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1][]

**（3）安装插件**

下载完成后，将其解压到 RabbitMQ 的 plugins 目录，如本人的目录路径为：D:\\Program Files\\RabbitMQ Server\\rabbitmq\_server-3.8.9\\plugins

打开一个新的cmd窗口（按快捷键“Win+R”，输入“cmd”）。

如果系统已经配置 RabbitMQ 环境变量，则执行下面的命令进行安装。

**运行命令：**rabbitmq-plugins enable rabbitmq\_delayed\_message\_exchange

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2][]

# 2、实现RabbitMQ消息队列延迟功能 #

**【示例】**创建 SpringBoot 项目，实现 RabbitMQ 消息队列延迟功能。

**（1）使用Maven添加依赖文件**

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

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.4.1</version>
    </dependency>

**（2）RabbitMQ的配置**

在 application.yml 配置文件中配置 RabbitMQ 信息：

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

**（3）RabbitMQ配置类**

创建 com.pjb.config 包，并创建 RabbitMqConfig 类（RabbitMQ配置类），配置交换机。

package com.pjb.config;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.CustomExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    
    /**
     * RabbitMQ配置类
     * @author pan_junbiao
     **/
    @Configuration
    public class RabbitMqConfig
    {
        public static final String DELAY_EXCHANGE_NAME = "delayed_exchange";
        public static final String DELAY_QUEUE_NAME = "delay_queue_name";
        public static final String DELAY_ROUTING_KEY = "delay_routing_key";
    
        @Bean
        public CustomExchange delayExchange()
        {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true, false, args);
        }
    
        @Bean
        public Queue queue()
        {
            Queue queue = new Queue(DELAY_QUEUE_NAME, true);
            return queue;
        }
    
        @Bean
        public Binding binding(Queue queue, CustomExchange delayExchange)
        {
            return BindingBuilder.bind(queue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
        }
    }

**（4）实现消息发送**

实现消息发送，这里设置消息延迟 5s。

package com.pjb.mq;
    
    import com.pjb.config.RabbitMqConfig;
    import org.springframework.amqp.AmqpException;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessagePostProcessor;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /**
     * 消息发送者
     * @author pan_junbiao
     **/
    @Service
    public class CustomSender
    {
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void sendMsg(String msg)
        {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("消息发送时间：" + sdf.format(new Date()));
            rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE_NAME, RabbitMqConfig.DELAY_ROUTING_KEY, msg, new MessagePostProcessor()
            {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException
                {
                    //消息延迟5秒
                    message.getMessageProperties().setHeader("x-delay", 5000);
                    return message;
                }
            });
        }
    }

**（5）实现消息接收**

package com.pjb.mq;
    
    import com.pjb.config.RabbitMqConfig;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    /**
     * 消息接收者
     * @author pan_junbiao
     **/
    @Component
    public class CustomReceiver
    {
        @RabbitListener(queues = RabbitMqConfig.DELAY_QUEUE_NAME)
        public void receive(String msg)
        {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println(sdf.format(new Date())+msg);
            System.out.println("Receiver：执行取消订单");
        }
    }

**（6）测试发送延迟消息**

package com.pjb;
    
    import com.pjb.mq.CustomSender;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    /**
     * RabbitMQ测试类
     * @author pan_junbiao
     **/
    @SpringBootTest
    public class MQTest
    {
        @Autowired
        private CustomSender customSender;
    
        @Test
        public void send() throws Exception
        {
            //发送消息
            customSender.sendMsg("支付超时，取消订单通知！");
    
            //程序延时15秒，否则程序立即执行完毕，则控制台无法看到消息队列延迟的结果
            Thread.sleep(15000);
        }
    }

**执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70]: /images/20210724/7bb557606e684b21a5a5bdab1fcf1bf6.png
[https_www.rabbitmq.com_community-plugins.html]: https://www.rabbitmq.com/community-plugins.html
[20210116160358308.jpg]: /images/20210724/0557e491805f46918693ca524205c2b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1]: /images/20210724/6fdf8839d0014cd28b998a9ebb33d517.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2]: /images/20210724/8fe491a01cd342f4b172c0c28e1f1614.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3]: /images/20210724/bc16025d41c44010a1a9ec2292b5eabc.png