RabbitMQ基础（2）——发布订阅/fanout模式 & topic模式 & rabbitmq回调确认 & 延迟队列（死信）设计

╰半橙微兮° 2023-10-14 17:11 10阅读 0赞

#### 目录 ####

*  引出
 *  点对点(simple)
 *  Work queues 一对多
 *  发布订阅/fanout模式
 *   *  以登陆验证码为例
     *  pom文件导包
     *  application.yml文件
     *  rabbitmq的配置
     *  生产者生成验证码，发送给交换机
     *  消费者消费验证码
 *  topic模式
 *   *  配置类增加配置
     *  生产者发送信息
     *  进行发送
     *  控制台查看
 *  rabbitmq回调确认
 *   *  配置类
     *  验证生产者发送是否成功
 *  延迟队列（死信）设计
 *   *  java代码步骤
     *   *  创建正常+死信队列
         *  配置类+常量
     *  生产者到正常队列
     *  消费者进行延迟消费
 *  延迟队列插件安装
 *   *  访问官网
     *  进入rabbitmq docker容器
     *  上传到linux服务器
     *  拷贝插件到容器中
     *  进入容器安装插件
     *  打开管理页面
 *  总结

## 引出 ##

--------------------

1.rabbitmq队列方式的梳理，点对点，一对多；  
2.发布订阅模式，交换机到消费者，以邮箱和手机验证码为例；  
3.topic模式，根据规则决定发送给哪个队列；  
4.rabbitmq回调确认，setConfirmCallback和setReturnsCallback；  
5.死信队列，延迟队列，创建方法，正常—死信，设置延迟时间；

## 点对点(simple) ##

> 点对对方式传输

![在这里插入图片描述][f68e35671eac4a01bb66104b1f37a859.png]

## Work queues 一对多 ##

> 1个生产者多个消费者

![在这里插入图片描述][887a0be918094f4b9329bd96d7899e14.png]

![在这里插入图片描述][32696a236ac94006b23379774826a036.png]

## 发布订阅/fanout模式 ##

> 生产者通过fanout扇出交换机群发消息给消费者，同一条消息每一个消费者都可以收到。

![在这里插入图片描述][b0b438aa88d549d8be8e5fff0a79a863.png]

### 以登陆验证码为例 ###

### pom文件导包 ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-mail</artifactId>
            </dependency>
    
            
            <dependency>
                <groupId>com.aliyun</groupId>
                <artifactId>aliyun-java-sdk-core</artifactId>
                <version>4.5.3</version>
            </dependency>
    
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

### application.yml文件 ###

server:
      port: 9099
    
    spring:
      # 模块的名字
      application:
        name: user-auth
    
    
      # 邮箱的配置
      mail:
        host: smtp.qq.com
        port: 587
        username: xxxx
        password: xxxxx
    
      # rabbitmq的配置
      rabbitmq:
        host: 192.168.111.130
        port: 5672
        username: admin
        password: 123
    
    logging:
      level:
        com.tianju.auth: debug

### rabbitmq的配置 ###

> 需要用到的常量

package com.tianju.auth.util;
    
    /**
     * rabbitmq的常量
     */
    public interface RabbitMqConstants {
        
        String MQ_MAIL_QUEUE="mq_email_queue";
        String MQ_PHONE_QUEUE="mq_phone_queue";
        String MQ_FANOUT_EXCHANGE="mq_fanout_exchange";
    
        // 参数 String name, boolean durable, boolean exclusive, boolean autoDelete
        boolean durable = true;
        boolean exclusive = false;
        boolean autoDelete = false;
    
    }

> RabbitMqConfig.java配置
> 
> 邮箱队列，电话队列，交换机；
> 
> 邮箱绑定交换机，电话绑定交换机；

创建队列参数说明：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>name</td> 
   <td>字符串值，queue的名称。</td> 
  </tr> 
  <tr> 
   <td>durable</td> 
   <td>布尔值，表示该 queue 是否持久化。 持久化意味着当 RabbitMQ 重启后，该 queue 是否会恢复/仍存在。 另外，需要注意的是，queue 的持久化不等于其中的消息也会被持久化。</td> 
  </tr> 
  <tr> 
   <td>exclusive</td> 
   <td>布尔值，表示该 queue 是否排它式使用。排它式使用意味着仅声明他的连接可见/可用，其它连接不可见/不可用。</td> 
  </tr> 
  <tr> 
   <td>autoDelete</td> 
   <td>布尔值，表示当该 queue 没“人”（connection）用时，是否会被自动删除。</td> 
  </tr> 
 </tbody> 
</table>

不指定 durable、exclusive 和 autoDelete 时，默认为 *true* 、 *false* 和 *false* 。表示持久化、非排它、不用自动删除。

创建交换机参数说明

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>name</td> 
   <td>字符串值，exchange 的名称。</td> 
  </tr> 
  <tr> 
   <td>durable</td> 
   <td>布尔值，表示该 exchage 是否持久化。 持久化意味着当 RabbitMQ 重启后，该 exchange 是否会恢复/仍存在。</td> 
  </tr> 
  <tr> 
   <td>autoDelete</td> 
   <td>布尔值，表示当该 exchange 没“人”（queue）用时，是否会被自动删除。</td> 
  </tr> 
 </tbody> 
</table>

> 不指定 durable 和 autoDelete 时，默认为 *`true`* 和 *`false`* 。表示持久化、不用自动删除

package com.tianju.auth.config;
    
    import com.tianju.auth.util.RabbitMqConstants;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.FanoutExchange;
    import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMqConfig {
        
    
        @Bean // 邮箱的队列
        public Queue mailQueue(){
        
            return new Queue(RabbitMqConstants.MQ_MAIL_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
    
        @Bean // 电话的队列
        public Queue phoneQueue(){
        
            return new Queue(RabbitMqConstants.MQ_PHONE_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
        @Bean // 交换机
        public FanoutExchange fanoutExchange(){
        
            return new FanoutExchange(RabbitMqConstants.MQ_FANOUT_EXCHANGE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.autoDelete);
        }
    
        @Bean
        public Binding mailBinding(){
        
            return BindingBuilder.bind(mailQueue())
                    .to(fanoutExchange());
        }
    
        @Bean
        public Binding phoneBinding(){
        
            return BindingBuilder.bind(phoneQueue())
                    .to(fanoutExchange());
        }
    
    }

### 生产者生成验证码，发送给交换机 ###

> 接口

package com.tianju.auth.service;
    
    public interface IUserService {
        
    
        /**
         * 生产者生成信息发送给交换机
         * @param msg 信息，这里是验证码
         */
        void sendCode(String msg);
    }

> 实现

package com.tianju.auth.service.impl;
    
    
    import com.tianju.auth.service.IUserService;
    import com.tianju.auth.util.RabbitMqConstants;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @Service
    @Slf4j
    public class UserServiceImpl implements IUserService {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Override
        public void sendCode(String msg) {
        
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_FANOUT_EXCHANGE,
                    "routingkey.fanout",
                    msg);
            log.debug("[生产者向交换机：] 发送一条信息：{}",msg);
        }
    
    }

> 测试类生成验证码，发给交换机

![在这里插入图片描述][60bd08eab3a54dc083f1462219e4ccd7.png]

package com.tianju.auth.service.impl;
    
    import cn.hutool.core.lang.Snowflake;
    import com.tianju.auth.service.IUserService;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    
    
    @SpringBootTest
    @RunWith(SpringJUnit4ClassRunner.class)
    
    
    public class UserServiceImplTest {
        
    
    
        @Autowired
        private IUserService userService;
    
        @Test
        public void sendCode() {
        
            String code = new Snowflake().nextIdStr().substring(0, 6);
            System.out.println(code);
            userService.sendCode(code);
        }
    }

### 消费者消费验证码 ###

package com.tianju.auth.consumer;
    
    
    import com.tianju.auth.service.IEmailService;
    import com.tianju.auth.util.RabbitMqConstants;
    import com.tianju.auth.util.SMSUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    @Slf4j
    @Service
    public class UserConsumer {
        
        @Autowired
        private IEmailService emailService;
    
        @RabbitListener(queues = RabbitMqConstants.MQ_MAIL_QUEUE)
        public void emailConsumer(String msg){
        
            log.debug("[email消费者：]消费{}",msg);
            emailService.sendEmail("xxxx@qq.com", "登陆验证码", msg);
        }
    
        @RabbitListener(queues = RabbitMqConstants.MQ_PHONE_QUEUE)
        public void phoneConsumer(String msg){
        
            log.debug("[phone消费者：]消费{}",msg);
            SMSUtil.send("xxxx", msg);
        }
    
    }

![在这里插入图片描述][ded10b5c9ff7438ba9e9b840ef547fcd.png]

## topic模式 ##

![在这里插入图片描述][2585b426376e4860be43576eada8088a.png]

> 例如： routingkey: my.orange.rabbit —-> Q1,Q2

![在这里插入图片描述][5bbb30c09947481a95cb66771c21ccd1.png]

### 配置类增加配置 ###

package com.tianju.auth.util;
    
    /**
     * rabbitmq的常量
     */
    public interface RabbitMqConstants {
        
        String MQ_MAIL_QUEUE="mq_email_queue";
        String MQ_PHONE_QUEUE="mq_phone_queue";
        String MQ_FANOUT_EXCHANGE="mq_fanout_exchange";
        
        String MQ_TOPIC_EXCHANGE="mq_topic_exchange";
    
        String MQ_TOPIC_QUEUE_A = "mq_topic_queue_a";
        String MQ_TOPIC_QUEUE_B = "mq_topic_queue_b";
    
        // 参数 String name, boolean durable, boolean exclusive, boolean autoDelete
        boolean durable = true;
        boolean exclusive = false;
        boolean autoDelete = false;
    
    }

package com.tianju.auth.config;
    
    import com.tianju.auth.util.RabbitMqConstants;
    import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class RabbitMqConfig {
        
    
        @Bean // 邮箱的队列
        public Queue mailQueue(){
        
            return new Queue(RabbitMqConstants.MQ_MAIL_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
    
        @Bean // 电话的队列
        public Queue phoneQueue(){
        
            return new Queue(RabbitMqConstants.MQ_PHONE_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
        @Bean // 交换机
        public FanoutExchange fanoutExchange(){
        
            return new FanoutExchange(RabbitMqConstants.MQ_FANOUT_EXCHANGE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.autoDelete);
        }
    
        @Bean
        public Binding mailBinding(){
        
            return BindingBuilder.bind(mailQueue())
                    .to(fanoutExchange());
        }
    
        @Bean
        public Binding phoneBinding(){
        
            return BindingBuilder.bind(phoneQueue())
                    .to(fanoutExchange());
        }
    
    
    
        @Bean // A队列
        public Queue topicAQueue(){
        
            return new Queue(RabbitMqConstants.MQ_TOPIC_QUEUE_A,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
    
        /**
         * topic模式相关配置
         */
    
        @Bean // B队列
        public Queue topicBQueue(){
        
            return new Queue(RabbitMqConstants.MQ_TOPIC_QUEUE_B,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
    
        @Bean // topic的交换机
        public TopicExchange topicMyExchange(){
        
            return new TopicExchange(RabbitMqConstants.MQ_TOPIC_EXCHANGE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.autoDelete);
        }
    
    
        @Bean
        public Binding topicAQueueBinding(){
        
            return BindingBuilder
                    .bind(topicAQueue())
                    .to(topicMyExchange())
                    .with("topic.xxx"); // 规则 topic.xxx
        }
    
        @Bean
        public Binding topicBQueueBinding(){
        
            return BindingBuilder
                    .bind(topicBQueue())
                    .to(topicMyExchange())
                    .with("topic.*"); // 规则 topic.xxx
        }
    
    }

### 生产者发送信息 ###

![在这里插入图片描述][1481ecc9efaf41c091714b82c1bc347c.png]

/**
         * topic模式下，生产者发送信息给交换机，可以决定给哪个队列发信息
         * @param msg 发送的信息
         * @param routingKey 类似正则表达式，决定给谁发
         *                   .with("topic.xxx"); // 规则 topic.xxx ---- A队列
         *                   .with("topic.*"); // 规则 topic.xxx   ---- B队列
         *                   在配置类中，如上所述配置，则如果输入的routingKey为 topic.xxx则给A和B发；
         *                                      如果输入的routingKey为 topic.yyy 则 只给B队列发；
         */
        void sendMsg(String msg,String routingKey);

> 实现

package com.tianju.auth.service.impl;
    
    import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
    import com.tianju.auth.entity.UserPrivs;
    import com.tianju.auth.mapper.UserMapper;
    import com.tianju.auth.service.IUserService;
    import com.tianju.auth.util.RabbitMqConstants;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.ArrayList;
    import java.util.List;
    
    @Service
    @Slf4j
    public class UserServiceImpl implements IUserService {
        
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Override
        public void sendCode(String msg) {
        
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_FANOUT_EXCHANGE,
                    "routingkey.fanout",
                    msg);
            log.debug("[生产者向交换机：] 发送一条信息：{}",msg);
        }
    
        @Override
        public void sendMsg(String msg,String routingKey) {
        
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_TOPIC_EXCHANGE,
                    routingKey, // "topic.yyy",此时只有B队列有信息
                    msg);
            log.debug("[生产者向交换机：] 发送一条信息：{}",msg);
        }
    
    
    }

### 进行发送 ###

package com.tianju.auth.service.impl;
    
    import cn.hutool.core.lang.Snowflake;
    import com.tianju.auth.service.IUserService;
    
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    
    
    @SpringBootTest
    @RunWith(SpringJUnit4ClassRunner.class)
    
    
    public class UserServiceImplTest {
        
    
        @Autowired
        private IUserService userService;
    
        @Test
        public void sendCode() {
        
            String code = new Snowflake().nextIdStr().substring(0, 6);
            System.out.println(code);
            userService.sendCode(code);
        }
    
        @Test
        public void sendTopic() {
        
            String code = new Snowflake().nextIdStr().substring(0, 6);
            System.out.println(code);
            userService.sendMsg(code,"topic.yyy");
        }
    }

![在这里插入图片描述][dc40f7009a5e4067b60957ebebb4483b.png]

### 控制台查看 ###

![在这里插入图片描述][214dd8bb1a8549ec9dd99fe23bb9586c.png]

## rabbitmq回调确认 ##

### 配置类 ###

spring:
      # rabbitmq的配置
      rabbitmq:
        host: 192.168.111.130
        port: 5672
        username: admin
        password: 123
        # 确认收到
        publisher-confirm-type: correlated
        publisher-returns: true

### 验证生产者发送是否成功 ###

> 使用RabbitTemplate的回调方法。
> 
> 先设置
> 
>  *  setConfirmCallback
>  *  setReturnsCallback

![在这里插入图片描述][29c69a9533834a688520fa380ca3d84d.png]

@Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Override
        public void sendCode(String msg) {
        
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_FANOUT_EXCHANGE,
                    "routingkey.fanout",
                    msg);
            log.debug("[生产者向交换机：] 发送一条信息：{}",msg);
        }
    
        @Override
        public void sendMsg(String msg,String routingKey) {
        
    
            // 如果发到交换机，看一下有没有反馈
            rabbitTemplate.setConfirmCallback((c,ack,message)->{
        
                log.debug("***** setConfirmCallback：ack--{}", ack); // 是否发送到交换机
                log.debug("***** setConfirmCallback：c-->{}",c);
                // channel error; protocol method: #method<channel.close>(reply-code=404,
                // reply-text=NOT_FOUND - no exchange 'aaaa' in vhost '/', class-id=60, method-id=40)
                log.debug("***** setConfirmCallback：m-->{}",message);
                if (ack){
        
                    log.debug("[生产者：] 发送信息到交换机{}","RabbitMqConstants.MQ_TOPIC_EXCHANGE");
                }else {
        
                    log.debug(message);
                }
            });
    
            rabbitTemplate.setReturnsCallback(r->{
        
                log.debug("返回文字{}", r.getReplyText());
                log.debug("返回code{}", r.getReplyCode());
                log.debug("返回Exchange{}", r.getExchange());
                log.debug("返回RoutingKey{}", r.getRoutingKey());
            });
    
    
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_TOPIC_EXCHANGE,
    //                "aaaa",// 失败的情况
                    routingKey, // "topic.yyy",此时只有B队列有信息
                    msg);
    
            log.debug("[生产者向交换机：] 发送一条信息：{}",msg);
        }

![在这里插入图片描述][add6f272ff754d5aa96bc788bc0f9b18.png]

rabbitTemplate.setConfirmCallback((c,ack,message)->{
        
        log.debug("******* setConfirmCallback:ack->{}",ack);
        log.debug("******* setConfirmCallback:c->{}",c);
        log.debug("******* setConfirmCallback:chanel->{}",message);
        if(ack){
        
            log.debug("[生产者]发送信息到达交换机{}","RabbitMqConstants.MQ_TOPIC_EXCHANGE");
        }else {
        
            log.debug(message);
        }
    });
    rabbitTemplate.setReturnsCallback(r->{
        
        log.debug("返回文字:{}",r.getReplyText());
        log.debug("返回code:{}",r.getReplyCode());
        log.debug("返回Exchange:{}",r.getExchange());
        log.debug("返回RoutingKey:{}",r.getRoutingKey());
    });
    rabbitTemplate.convertAndSend(
            RabbitMqConstants.MQ_TOPIC_EXCHANGE,
            "abc.xxx",
            msg
    );

@Test
        public void sendTopic() {
        
            String code = new Snowflake().nextIdStr().substring(0, 6);
            System.out.println(code);
            userService.sendMsg(code,"topic.rrr");
        }

## 延迟队列（死信）设计 ##

[Documentation: Table of Contents — RabbitMQ][Documentation_ Table of Contents _ RabbitMQ]

![在这里插入图片描述][8d012596a93d45acaf606bcef7dde082.png]

![在这里插入图片描述][15d187f391624fa6aa3905df97230d56.png]

### java代码步骤 ###

#### 创建正常+死信队列 ####

package com.tianju.mq.config;
    
    import com.tianju.mq.constants.RabbitMqConstants;
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.DirectExchange;
    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;
    
    @Configuration
    public class RabbitMqConfig {
        
    
        @Bean
        public DirectExchange normalExchange(){
        
            return new DirectExchange(RabbitMqConstants.MQ_NORMAL_EXCHANGE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.autoDelete);
        }
        @Bean
        public Queue normalQueue(){
        
            Map<String, Object> map = new HashMap<>(2);
            map.put("x-dead-letter-exchange",RabbitMqConstants.MQ_DELAY_EXCHANGE);
            map.put("x-dead-letter-routing-key",RabbitMqConstants.MQ_DELAY_ROUTING_KEY);
            return new Queue(
                    RabbitMqConstants.MQ_NORMAL_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete,
                    map);
        }
        @Bean
        public Binding normalBinding(){
        
            return BindingBuilder.bind(normalQueue())
                    .to(normalExchange())
                    .with(RabbitMqConstants.MQ_NORMAL_ROUTING_KEY);
        }
    
    
        //------------------死信队列设计--------------------------
        /**
         * 死信(延迟)队列
         * @return
         */
        @Bean
        public Queue delayQueue(){
        
            return new Queue(RabbitMqConstants.MQ_DELAY_QUEUE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.exclusive,
                    RabbitMqConstants.autoDelete);
        }
        /**
         * 死信交换机
         * @return
         */
        @Bean
        public DirectExchange delayExchange(){
        
            return new DirectExchange(RabbitMqConstants.MQ_DELAY_EXCHANGE,
                    RabbitMqConstants.durable,
                    RabbitMqConstants.autoDelete);
        }
        /**
         * 死信交换机队列绑定
         * @return
         */
        @Bean
        public Binding delayBinding(){
        
            return BindingBuilder.bind(delayQueue())
                    .to(delayExchange())
                    .with(RabbitMqConstants.MQ_DELAY_ROUTING_KEY);
        }
    }

#### 配置类+常量 ####

package com.tianju.mq.constants;
    
    public interface RabbitMqConstants {
        
        String MQ_DELAY_QUEUE = "mq_delay_queue"; // 延迟队列，死信队列
        String MQ_DELAY_EXCHANGE = "mq_delay_exchange"; // 死信交换机
        String MQ_DELAY_ROUTING_KEY = "mq_delay_routing_key"; // 死信路由
    
        // 正常的队列，交换机，路由
        String MQ_NORMAL_QUEUE = "mq_normal_queue";
        String MQ_NORMAL_EXCHANGE = "mq_normal_exchange";
        String MQ_NORMAL_ROUTING_KEY = "mq_normal_routing_key";
    
        // 参数
        boolean durable = true;
        boolean exclusive = false;
        boolean autoDelete = false;
    }

server:
      port: 9099
    
    spring:
    
      # 邮箱的配置
      mail:
        host: smtp.qq.com
        port: 587
        username: xxxxx.com
        password: xxxxx
    
      # rabbitmq的配置
      rabbitmq:
        host: 192.168.111.130
        port: 5672
        username: admin
        password: 123
        # 确认收到
        publisher-confirm-type: correlated
        publisher-returns: true
    
    
    logging:
      level:
        com.tianju.mq: debug

### 生产者到正常队列 ###

package com.tianju.mq.service;
    
    public interface IUserService {
        
        /**
         * 延迟队列的生产者
         * @param msg 发送的信息
         * @param delayTime 延迟的时间，毫秒
         */
        void sendDelay(String msg,int delayTime);
    }

package com.tianju.mq.service.impl;
    
    import com.tianju.mq.constants.RabbitMqConstants;
    import com.tianju.mq.service.IUserService;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.MessageProperties;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    
    import java.util.Date;
    
    @Service
    @Slf4j
    public class UserServiceImpl implements IUserService {
        
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Override
        public void sendDelay(String msg, int delayTime) {
        
            rabbitTemplate.convertAndSend(
                    RabbitMqConstants.MQ_NORMAL_EXCHANGE,
                    RabbitMqConstants.MQ_NORMAL_ROUTING_KEY,
                    msg,
                    process->{
        
                        process.getMessageProperties().setExpiration(String.valueOf(delayTime));
                        return process;
                    }
            );
            log.debug("[生产者：]发送消息:{},时间{},延迟{}秒",msg,new Date(),delayTime/1000);
        }
    }

![在这里插入图片描述][1b1daeba52f5415594493e4840c6f07e.png]

### 消费者进行延迟消费 ###

package com.tianju.mq.consumer;
    
    import com.tianju.mq.constants.RabbitMqConstants;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    
    @Service
    @Slf4j
    public class UserConsumer {
        
    
        @RabbitListener(queues = RabbitMqConstants.MQ_DELAY_QUEUE)
        public void delayConsume(String msg){
        
            log.debug("[消费者消费信息:{},时间:{}",msg,new Date());
        }
    }

![在这里插入图片描述][620e3e1a7ca347ae99609d4e4007833f.png]

## 延迟队列插件安装 ##

### 访问官网 ###

[Community Plugins — RabbitMQ][Community Plugins _ RabbitMQ]

![在这里插入图片描述][7fcef17c83154354abc72650c7ac47f6.png]

![在这里插入图片描述][0af4e0bce3514cef8dccc967760af259.png]

### 进入rabbitmq docker容器 ###

[root@localhost ~]# docker exec -it rabbitmq bash

查询插件列表是否存在延迟插件

root@6d2342d51b11:/plugins# rabbitmq-plugins list
    Listing plugins with pattern ".*" ...
     Configured: E = explicitly enabled; e = implicitly enabled
     | Status: * = running on rabbit@6d2342d51b11
     |/
    [  ] rabbitmq_amqp1_0                  3.9.11
    [  ] rabbitmq_auth_backend_cache       3.9.11
    [  ] rabbitmq_auth_backend_http        3.9.11
    [  ] rabbitmq_auth_backend_ldap        3.9.11
    [  ] rabbitmq_auth_backend_oauth2      3.9.11
    [  ] rabbitmq_auth_mechanism_ssl       3.9.11
    [  ] rabbitmq_consistent_hash_exchange 3.9.11
    [  ] rabbitmq_event_exchange           3.9.11
    [  ] rabbitmq_federation               3.9.11
    [  ] rabbitmq_federation_management    3.9.11
    [  ] rabbitmq_jms_topic_exchange       3.9.11
    [E*] rabbitmq_management               3.9.11
    [e*] rabbitmq_management_agent         3.9.11
    [  ] rabbitmq_mqtt                     3.9.11
    [  ] rabbitmq_peer_discovery_aws       3.9.11
    [  ] rabbitmq_peer_discovery_common    3.9.11
    [  ] rabbitmq_peer_discovery_consul    3.9.11
    [  ] rabbitmq_peer_discovery_etcd      3.9.11
    [  ] rabbitmq_peer_discovery_k8s       3.9.11
    [E*] rabbitmq_prometheus               3.9.11
    [  ] rabbitmq_random_exchange          3.9.11
    [  ] rabbitmq_recent_history_exchange  3.9.11
    [  ] rabbitmq_sharding                 3.9.11
    [  ] rabbitmq_shovel                   3.9.11
    [  ] rabbitmq_shovel_management        3.9.11
    [  ] rabbitmq_stomp                    3.9.11
    [  ] rabbitmq_stream                   3.9.11
    [  ] rabbitmq_stream_management        3.9.11
    [  ] rabbitmq_top                      3.9.11
    [  ] rabbitmq_tracing                  3.9.11
    [  ] rabbitmq_trust_store              3.9.11
    [e*] rabbitmq_web_dispatch             3.9.11
    [  ] rabbitmq_web_mqtt                 3.9.11
    [  ] rabbitmq_web_mqtt_examples        3.9.11
    [  ] rabbitmq_web_stomp                3.9.11
    [  ] rabbitmq_web_stomp_examples       3.9.11

> 下载支持3.9.x的插件

[https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases?after=rabbitmq\_v3\_6\_12][https_github.com_rabbitmq_rabbitmq-delayed-message-exchange_releases_after_rabbitmq_v3_6_12]

![在这里插入图片描述][e4664d143bb745b3bddfb04a47f52b4c.png]

![在这里插入图片描述][f85bf258563c4d289a4db0a515f57355.png]

退出容器：

root@6d2342d51b11:/plugins# exit
    exit

### 上传到linux服务器 ###

在/usr/local/software/下创建文件夹rabbitmq/plugins

[root@localhost software]# mkdir -p rabbitmq/plugins

![在这里插入图片描述][54f05fac7653405f9b73460a69dcf49b.png]

### 拷贝插件到容器中 ###

[root@localhost plugins]# docker cp ./rabbitmq_delayed_message_exchange-3.9.0.ez rabbitmq:/plugins

### 进入容器安装插件 ###

[root@localhost plugins]# docker  exec -it rabbitmq bash
    root@6d2342d51b11:/# rabbitmq-plugins enable rabbitmq_delayed_message_exchange

![在这里插入图片描述][438cf30d63de4a2d8cdd99829b763f64.png]

### 打开管理页面 ###

> 进入Exchange页面，下拉Type看是否已经安装成功。

![在这里插入图片描述][87c959fe4b7647948118bb30addc2a74.png]

--------------------

## 总结 ##

[f68e35671eac4a01bb66104b1f37a859.png]: https://img-blog.csdnimg.cn/f68e35671eac4a01bb66104b1f37a859.png
[887a0be918094f4b9329bd96d7899e14.png]: https://img-blog.csdnimg.cn/887a0be918094f4b9329bd96d7899e14.png
[32696a236ac94006b23379774826a036.png]: https://img-blog.csdnimg.cn/32696a236ac94006b23379774826a036.png
[b0b438aa88d549d8be8e5fff0a79a863.png]: https://img-blog.csdnimg.cn/b0b438aa88d549d8be8e5fff0a79a863.png
[60bd08eab3a54dc083f1462219e4ccd7.png]: https://img-blog.csdnimg.cn/60bd08eab3a54dc083f1462219e4ccd7.png
[ded10b5c9ff7438ba9e9b840ef547fcd.png]: https://img-blog.csdnimg.cn/ded10b5c9ff7438ba9e9b840ef547fcd.png
[2585b426376e4860be43576eada8088a.png]: https://img-blog.csdnimg.cn/2585b426376e4860be43576eada8088a.png
[5bbb30c09947481a95cb66771c21ccd1.png]: https://img-blog.csdnimg.cn/5bbb30c09947481a95cb66771c21ccd1.png
[1481ecc9efaf41c091714b82c1bc347c.png]: https://img-blog.csdnimg.cn/1481ecc9efaf41c091714b82c1bc347c.png
[dc40f7009a5e4067b60957ebebb4483b.png]: https://img-blog.csdnimg.cn/dc40f7009a5e4067b60957ebebb4483b.png
[214dd8bb1a8549ec9dd99fe23bb9586c.png]: https://img-blog.csdnimg.cn/214dd8bb1a8549ec9dd99fe23bb9586c.png
[29c69a9533834a688520fa380ca3d84d.png]: https://img-blog.csdnimg.cn/29c69a9533834a688520fa380ca3d84d.png
[add6f272ff754d5aa96bc788bc0f9b18.png]: https://img-blog.csdnimg.cn/add6f272ff754d5aa96bc788bc0f9b18.png
[Documentation_ Table of Contents _ RabbitMQ]: https://www.rabbitmq.com/documentation.html
[8d012596a93d45acaf606bcef7dde082.png]: https://img-blog.csdnimg.cn/8d012596a93d45acaf606bcef7dde082.png
[15d187f391624fa6aa3905df97230d56.png]: https://img-blog.csdnimg.cn/15d187f391624fa6aa3905df97230d56.png
[1b1daeba52f5415594493e4840c6f07e.png]: https://img-blog.csdnimg.cn/1b1daeba52f5415594493e4840c6f07e.png
[620e3e1a7ca347ae99609d4e4007833f.png]: https://img-blog.csdnimg.cn/620e3e1a7ca347ae99609d4e4007833f.png
[Community Plugins _ RabbitMQ]: https://www.rabbitmq.com/community-plugins.html
[7fcef17c83154354abc72650c7ac47f6.png]: https://img-blog.csdnimg.cn/7fcef17c83154354abc72650c7ac47f6.png
[0af4e0bce3514cef8dccc967760af259.png]: https://img-blog.csdnimg.cn/0af4e0bce3514cef8dccc967760af259.png
[https_github.com_rabbitmq_rabbitmq-delayed-message-exchange_releases_after_rabbitmq_v3_6_12]: https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases?after=rabbitmq_v3_6_12
[e4664d143bb745b3bddfb04a47f52b4c.png]: https://img-blog.csdnimg.cn/e4664d143bb745b3bddfb04a47f52b4c.png
[f85bf258563c4d289a4db0a515f57355.png]: https://img-blog.csdnimg.cn/f85bf258563c4d289a4db0a515f57355.png
[54f05fac7653405f9b73460a69dcf49b.png]: https://img-blog.csdnimg.cn/54f05fac7653405f9b73460a69dcf49b.png
[438cf30d63de4a2d8cdd99829b763f64.png]: https://img-blog.csdnimg.cn/438cf30d63de4a2d8cdd99829b763f64.png
[87c959fe4b7647948118bb30addc2a74.png]: https://img-blog.csdnimg.cn/87c959fe4b7647948118bb30addc2a74.png