SpringBoot整合RabbitMQ实现消息的发送与接收

傷城~ 2021-09-07 06:06 579阅读 0赞

RabbitMQ 是采用 Erlang 语言实现 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）的消息中间件，用于在分布式系统中存储转发消息。RabbitMQ 凭借其高可靠、易扩展、高可用及丰富的功能特性收到越来越多企业的青睐。

**【实例】**SpringBoot整合RabbitMQ实现消息的发送与接收。

**实例要求：**

1.  实现SpringBoot整合RabbitMQ框架。
2.  实现RabbitMQ消息确认机制（ACK）。
3.  实现RabbitMQ消息队列延迟功能。
4.  实现RabbitMQ重试机制。
5.  实现发送与接收JSON、Map格式数据。

> **相关技术：**
> 
> [《RabbitMQ交换器类型与消息的发送/接收模式》][RabbitMQ]
> 
> [《RabbitMQ消息确认机制（ACK）》][RabbitMQ_ACK]
> 
> [《RabbitMQ实现消息队列延迟功能》][RabbitMQ 1]
> 
> [《RabbitMQ重试机制》][RabbitMQ 2]
> 
> [《RabbitMQ实现JSON、Map格式数据的发送与接收》][RabbitMQ_JSON_Map]
> 
> [《SpringBoot整合RabbitMQ实现消息的发送与接收》][SpringBoot_RabbitMQ]

# 1、消息发送端 #

消息发送端实现消息的发送、消息确认功能（交换器确认、队列确认）。

## 1.1 创建项目 ##

创建第一个 SpringBoot 项目（ RabbitmqProvider 消息发送项目），并整合 RabbitMQ 框架。项目结构如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70][]

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

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

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

spring:
      # 项目名称
      application:
        name: rabbitmq-provider
      # RabbitMQ服务配置
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest
        password: guest
        # 消息确认（ACK）
        publisher-confirm-type: correlated #确认消息已发送到交换机(Exchange)
        publisher-returns: true #确认消息已发送到队列(Queue)

## 1.2 配置类（Config层） ##

在com.pjb.config包中，创建RabbitMqConfig类（RabbitMQ配置类），代码如下：

package com.pjb.config;
    
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    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 DIRECT_QUEUE = "direct_queue"; //Direct队列名称
        public static final String DIRECT_EXCHANGE = "direct_exchange"; //交换器名称
        public static final String DIRECT_ROUTING_KEY = "direct_routing_key"; //路由键
    
        public static final String DELAY_QUEUE = "delay_queue"; //延时队列名称
        public static final String DELAY_EXCHANGE = "delay_exchange"; //交换器名称
        public static final String DELAY_ROUTING_KEY = "delay_routing_key"; //路由键
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
    
        @Bean
        public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory)
        {
            RabbitTemplate rabbitTemplate = new RabbitTemplate();
            rabbitTemplate.setConnectionFactory(connectionFactory);
    
            //设置Json转换器
            rabbitTemplate.setMessageConverter(jsonMessageConverter());
    
            //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
            rabbitTemplate.setMandatory(true);
    
            //确认消息送到交换机(Exchange)回调
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback()
            {
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause)
                {
                    System.out.println("\n确认消息送到交换机(Exchange)结果：");
                    System.out.println("相关数据：" + correlationData);
                    System.out.println("是否成功：" + ack);
                    System.out.println("错误原因：" + cause);
                }
            });
    
            //确认消息送到队列(Queue)回调
            rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback()
            {
                @Override
                public void returnedMessage(ReturnedMessage returnedMessage)
                {
                    System.out.println("\n确认消息送到队列(Queue)结果：");
                    System.out.println("发生消息：" + returnedMessage.getMessage());
                    System.out.println("回应码：" + returnedMessage.getReplyCode());
                    System.out.println("回应信息：" + returnedMessage.getReplyText());
                    System.out.println("交换机：" + returnedMessage.getExchange());
                    System.out.println("路由键：" + returnedMessage.getRoutingKey());
                }
            });
            return rabbitTemplate;
        }
    
        /**
         * Json转换器
         */
        @Bean
        public Jackson2JsonMessageConverter jsonMessageConverter()
        {
            return new Jackson2JsonMessageConverter();
        }
    
        /**
         * Direct交换器
         */
        @Bean
        public DirectExchange directExchange()
        {
            /**
             * 创建交换器，参数说明：
             * String name：交换器名称
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             */
            return new DirectExchange(DIRECT_EXCHANGE, true, false);
        }
    
        /**
         * 队列
         */
        @Bean
        public Queue directQueue()
        {
            /**
             * 创建队列，参数说明：
             * String name：队列名称。
             * boolean durable：设置是否持久化，默认是 false。durable 设置为 true 表示持久化，反之是非持久化。
             * 持久化的队列会存盘，在服务器重启的时候不会丢失相关信息。
             * boolean exclusive：设置是否排他，默认也是 false。为 true 则设置队列为排他。
             * boolean autoDelete：设置是否自动删除，为 true 则设置队列为自动删除，
             * 当没有生产者或者消费者使用此队列，该队列会自动删除。
             * Map<String, Object> arguments：设置队列的其他一些参数。
             */
            return new Queue(DIRECT_QUEUE, true, false, false, null);
        }
    
        /**
         * 绑定
         */
        @Bean
        Binding directBinding(DirectExchange directExchange, Queue directQueue)
        {
            //将队列和交换机绑定, 并设置用于匹配键：routingKey路由键
            return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);
        }
    
        /******************************延时队列******************************/
    
        @Bean
        public CustomExchange delayExchange()
        {
            Map<String, Object> args = new HashMap<>();
            args.put("x-delayed-type", "direct");
            return new CustomExchange(DELAY_EXCHANGE, "x-delayed-message", true, false, args);
        }
    
        @Bean
        public Queue delayQueue()
        {
            Queue queue = new Queue(DELAY_QUEUE, true);
            return queue;
        }
    
        @Bean
        public Binding delaybinding(Queue delayQueue, CustomExchange delayExchange)
        {
            return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
        }
    }

## 1.3 实体类（Entity层） ##

在com.pjb.entity包中，创建UserInfo类（用户信息实体类）。

package com.pjb.entity;
    
    /**
     * 用户信息实体类
     * @author pan_junbiao
     **/
    public class UserInfo
    {
        private int userId; //用户编号
        private String userName; //用户姓名
        private String blogUrl; //博客地址
        private String blogRemark; //博客信息
    
        //省略getter与setter方法...
    }

## 1.4 消息发送层（Sender层） ##

在com.pjb.sender包下，创建UserSender接口（用户消息发送服务接口）。

package com.pjb.sender;
    
    import com.pjb.entity.UserInfo;
    
    import java.util.Map;
    
    /**
     * 用户消息发送服务接口
     * @author pan_junbiao
     **/
    public interface UserSender
    {
        /**
         * 发送用户信息Json格式数据
         * @param userInfo 用户信息实体类
         */
        public void sendUserJson(UserInfo userInfo);
    
        /**
         * 延时发送用户信息Map格式数据
         * @param userMap 用户信息Map
         */
        public void sendDelayUserMap(Map userMap);
    }

在com.pjb.sender.impl包下，创建UserSenderImpl类（用户消息发送服务类）。

package com.pjb.sender.impl;
    
    import com.pjb.config.RabbitMqConfig;
    import com.pjb.entity.UserInfo;
    import com.pjb.sender.UserSender;
    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;
    import java.util.Map;
    
    /**
     * 用户消息发送服务类
     * @author pan_junbiao
     **/
    @Service
    public class UserSenderImpl implements UserSender
    {
        @Autowired
        RabbitTemplate rabbitTemplate;
    
        //时间格式
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        /**
         * 发送用户信息Json格式数据
         * @param userInfo 用户信息实体类
         */
        @Override
        public void sendUserJson(UserInfo userInfo)
        {
            /**
             * 发送消息，参数说明：
             * String exchange：交换器名称。
             * String routingKey：路由键。
             * Object object：发送内容。
             */
            rabbitTemplate.convertAndSend(RabbitMqConfig.DIRECT_EXCHANGE, RabbitMqConfig.DIRECT_ROUTING_KEY, userInfo);
    
            System.out.println("Json格式数据消息发送成功，发送时间：" + dateFormat.format(new Date()));
        }
    
        /**
         * 延时发送用户信息Map格式数据
         * @param userMap 用户信息Map
         */
        @Override
        public void sendDelayUserMap(Map userMap)
        {
            rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE, RabbitMqConfig.DELAY_ROUTING_KEY, userMap, new MessagePostProcessor()
            {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException
                {
                    //消息延迟5秒
                    message.getMessageProperties().setHeader("x-delay", 5000);
                    return message;
                }
            });
    
            System.out.println("Map格式数据消息发送成功，发送时间：" + dateFormat.format(new Date()));
        }
    }

# 2、消息接收端 #

消息接收端实现消息的接收、消息接收确认功能。

## 2.1 创建项目 ##

创建第二个 SpringBoot 项目（ RabbitmqConsumer 消息接收项目），并整合 RabbitMQ 框架。项目结构如下图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1][]

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

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

在 application.yml 配置文件中配置 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.2 配置类（Config层） ##

在com.pjb.config包中，创建RabbitMqConfig类（RabbitMQ配置类），代码如下：

package com.pjb.config;
    
    import com.pjb.receiver.impl.AckReceiver;
    import org.springframework.amqp.core.AcknowledgeMode;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    /**
     * RabbitMQ配置类
     * @author pan_junbiao
     **/
    @Configuration
    public class RabbitMqConfig
    {
        public static final String DIRECT_QUEUE = "direct_queue"; //Direct队列名称
        public static final String DELAY_QUEUE = "delay_queue"; //延时队列名称
    
        /**
         * 消息接收确认处理类
         */
        @Autowired
        private AckReceiver ackReceiver;
    
        @Autowired
        private CachingConnectionFactory connectionFactory;
    
        /**
         * 客户端配置
         * 配置手动确认消息、消息接收确认
         */
        @Bean
        public SimpleMessageListenerContainer simpleMessageListenerContainer()
        {
            //消费者数量，默认10
            int DEFAULT_CONCURRENT = 10;
    
            //每个消费者获取最大投递数量 默认50
            int DEFAULT_PREFETCH_COUNT = 50;
    
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
            container.setConcurrentConsumers(DEFAULT_CONCURRENT);
            container.setMaxConcurrentConsumers(DEFAULT_PREFETCH_COUNT);
    
            // RabbitMQ默认是自动确认，这里改为手动确认消息
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    
            //添加队列，可添加多个队列
            container.addQueues(new Queue(DIRECT_QUEUE,true));
            container.addQueues(new Queue(DELAY_QUEUE,true));
    
            //设置消息处理类
            container.setMessageListener(ackReceiver);
    
            return container;
        }
    }

## 2.3 实体类（Entity层） ##

在com.pjb.entity包中，创建UserInfo类（用户信息实体类）。

## 1.4 消息接收层（Receiver层） ##

在com.pjb.receiver包下，创建UserReceiver接口（用户消息接收接口）。

package com.pjb.receiver;
    
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    
    /**
     * 用户消息接收接口
     * @author pan_junbiao
     **/
    public interface UserReceiver
    {
        /**
         * 接收用户信息Json格式数据
         */
        public void receiverUserJson(Message message, Channel channel) throws Exception;
    
        /**
         * 延时接收用户信息Map格式数据
         */
        public void receiverDelayUserMap(Message message, Channel channel) throws Exception;
    }

在com.pjb.receiver.impl包下，创建UserReceiverImpl类（用户消息接收类）。

package com.pjb.receiver.impl;
    
    import com.fasterxml.jackson.databind.JavaType;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.pjb.entity.UserInfo;
    import com.pjb.receiver.UserReceiver;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.stereotype.Service;
    
    import java.util.Map;
    
    /**
     * 用户消息接收类
     * @author pan_junbiao
     **/
    @Service
    public class UserReceiverImpl implements UserReceiver
    {
        /**
         * 接收用户信息Json格式数据
         */
        @Override
        public void receiverUserJson(Message message, Channel channel) throws Exception
        {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try
            {
                //将JSON格式数据转换为实体对象
                ObjectMapper mapper = new ObjectMapper();
                UserInfo userInfo = mapper.readValue(message.getBody(), UserInfo.class);
    
                System.out.println("接收者收到JSON格式消息：");
                System.out.println("用户编号：" + userInfo.getUserId());
                System.out.println("用户名称：" + userInfo.getUserName());
                System.out.println("博客地址：" + userInfo.getBlogUrl());
                System.out.println("博客信息：" + userInfo.getBlogRemark());
    
                /**
                 * 确认消息，参数说明：
                 * long deliveryTag：唯一标识 ID。
                 * boolean multiple：是否批处理，当该参数为 true 时，
                 * 则可以一次性确认 deliveryTag 小于等于传入值的所有消息。
                 */
                channel.basicAck(deliveryTag, true);
    
                /**
                 * 否定消息，参数说明：
                 * long deliveryTag：唯一标识 ID。
                 * boolean multiple：是否批处理，当该参数为 true 时，
                 * 则可以一次性确认 deliveryTag 小于等于传入值的所有消息。
                 * boolean requeue：如果 requeue 参数设置为 true，
                 * 则 RabbitMQ 会重新将这条消息存入队列，以便发送给下一个订阅的消费者；
                 * 如果 requeue 参数设置为 false，则 RabbitMQ 立即会还把消息从队列中移除，
                 * 而不会把它发送给新的消费者。
                 */
                //channel.basicNack(deliveryTag, true, false);
            }
            catch (Exception e)
            {
                /**
                 * 拒绝消息，参数说明：
                 * long deliveryTag：唯一标识 ID。
                 * boolean requeue：如果 requeue 参数设置为 true，
                 * 则 RabbitMQ 会重新将这条消息存入队列，以便发送给下一个订阅的消费者；
                 * 如果 requeue 参数设置为 false，则 RabbitMQ 立即会还把消息从队列中移除，
                 * 而不会把它发送给新的消费者。
                 */
                channel.basicReject(deliveryTag, false);
    
                e.printStackTrace();
            }
        }
    
        /**
         * 延时接收用户信息Map格式数据
         */
        @Override
        public void receiverDelayUserMap(Message message, Channel channel) throws Exception
        {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try
            {
                //将JSON格式数据转换为Map对象
                ObjectMapper mapper = new ObjectMapper();
                JavaType javaType = mapper.getTypeFactory().constructMapType(Map.class, String.class, Object.class);
                Map<String, Object> resultMap = mapper.readValue(message.getBody(),javaType);
    
                System.out.println("接收者收到Map格式消息：");
                System.out.println("用户编号：" + resultMap.get("userId"));
                System.out.println("用户名称：" + resultMap.get("userName"));
                System.out.println("博客地址：" + resultMap.get("blogUrl"));
                System.out.println("博客信息：" + resultMap.get("userRemark"));
    
                //确认消息
                channel.basicAck(deliveryTag, true);
    
                //否定消息
                //channel.basicNack(deliveryTag, true, false);
            }
            catch (Exception e)
            {
                //拒绝消息
                channel.basicReject(deliveryTag, false);
    
                e.printStackTrace();
            }
        }
    }

在com.pjb.receiver.impl包下，创建AckReceiver类（消息接收确认处理类）。

package com.pjb.receiver.impl;
    
    import com.pjb.config.RabbitMqConfig;
    import com.pjb.receiver.UserReceiver;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    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 AckReceiver implements ChannelAwareMessageListener
    {
        /**
         * 用户消息接收类
         */
        @Autowired
        private UserReceiver userReceiver;
    
        @Override
        public void onMessage(Message message, Channel channel) throws Exception
        {
            //时间格式
            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.println("消息接收成功，接收时间：" + dateFormat.format(new Date()) + "\n");
    
            //获取队列名称
            String queueName = message.getMessageProperties().getConsumerQueue();
    
            //接收用户信息Json格式数据
            if (queueName.equals(RabbitMqConfig.DIRECT_QUEUE))
            {
                userReceiver.receiverUserJson(message, channel);
            }
    
            //延时接收用户信息Map格式数据
            if (queueName.equals(RabbitMqConfig.DELAY_QUEUE))
            {
                userReceiver.receiverDelayUserMap(message, channel);
            }
    
            //多个队列的处理，则如上述代码，继续添加方法....
        }
    }

# 3、运行测试 #

完成上述2个项目代码后，就可以进行运行测试了。首先启动第二个 SpringBoot 项目（ RabbitmqConsumer 消息接收项目），让其一直运行着。

## 3.1 消息队列的发送与接收 ##

在第一个 SpringBoot 项目（ RabbitmqProvider 消息发送项目），中创建测试类，并编写发送用户信息Json格式数据的方法。

package com.pjb.sender;
    
    import com.pjb.entity.UserInfo;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 用户消息发送测试类
     * @author pan_junbiao
     **/
    @SpringBootTest
    public class UserSenderTest
    {
        /**
         * 用户消息发送类
         */
        @Autowired
        private UserSender userSender;
    
        /**
         * 发送用户信息Json格式数据
         */
        @Test
        public void sendUserJson() throws Exception
        {
            //创建用户信息
            UserInfo userInfo = new UserInfo();
            userInfo.setUserId(1);
            userInfo.setUserName("pan_junbiao的博客");
            userInfo.setBlogUrl("https://blog.csdn.net/pan_junbiao");
            userInfo.setBlogRemark("您好，欢迎访问 pan_junbiao的博客");
    
            //执行发送
            userSender.sendUserJson(userInfo);
    
            //由于这里使用的是测试方法，当测试方法结束，RabbitMQ相关的资源也就关闭了，
            //会导致消息确认的回调出现问题，所有加段延时，该延时与业务无关
            Thread.sleep(2000);
        }
    }

**发送端执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2][]

**接收端执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3][]

## 3.2 消息队列的延迟功能 ##

编写代码，测试发送用户信息Map格式数据和消息队列的延迟功能。

/**
     * 用户消息发送类
     */
    @Autowired
    private UserSender userSender;
    
    /**
     * 延时发送用户信息Map格式数据
     */
    @Test
    public void sendDelayUserMap() throws Exception
    {
        //创建用户信息Map
        Map<String, Object> userMap = new HashMap<>();
        userMap.put("userId", "1");
        userMap.put("userName", "pan_junbiao的博客");
        userMap.put("blogUrl", "https://blog.csdn.net/pan_junbiao");
        userMap.put("userRemark", "您好，欢迎访问 pan_junbiao的博客");
    
        //执行发送
        userSender.sendDelayUserMap(userMap);
    
        //由于这里使用的是测试方法，当测试方法结束，RabbitMQ相关的资源也就关闭了，
        //会导致消息确认的回调出现问题，所有加段延时，该延时与业务无关
        Thread.sleep(2000);
    }

**发送端执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 4][]

**接收端执行结果：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 5][]

> **结果说明：**
> 
> 在代码1.4中，设置了消息的延时为5秒钟。从上图可以看出，消息的发送时间与消息的接收时间正好间隔5秒钟，说明 RabbitMQ 消息队列的延迟功能设置成功。

> **源代码下载：**[https://github.com/kevinpanjunbiao/RabbitMqDemo][https_github.com_kevinpanjunbiao_RabbitMqDemo]
> 
> **原文地址：**[https://blog.csdn.net/pan\_junbiao/article/details/113560849][SpringBoot_RabbitMQ]

[RabbitMQ]: https://blog.csdn.net/pan_junbiao/article/details/112838589
[RabbitMQ_ACK]: https://blog.csdn.net/pan_junbiao/article/details/112956537
[RabbitMQ 1]: https://blog.csdn.net/pan_junbiao/article/details/112711560
[RabbitMQ 2]: https://blog.csdn.net/pan_junbiao/article/details/113038421
[RabbitMQ_JSON_Map]: https://blog.csdn.net/pan_junbiao/article/details/113522993
[SpringBoot_RabbitMQ]: https://blog.csdn.net/pan_junbiao/article/details/113560849
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70]: /images/20210724/475c4ca9fefd434894cc8515c65aa94b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 1]: /images/20210724/86639173943e4ea581d23076c08625e5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 2]: /images/20210724/eddd768e997c4aaca0bb718aff2ad7ba.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 3]: /images/20210724/760fbc73319d49fca43c1ca95a86e3d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 4]: /images/20210724/b90bb62d2a13444a85d09f7ceb71609c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3Bhbl9qdW5iaWFv_size_16_color_FFFFFF_t_70 5]: /images/20210724/60f34552b5a04989939eb15023007743.png
[https_github.com_kevinpanjunbiao_RabbitMqDemo]: https://github.com/kevinpanjunbiao/RabbitMqDemo