延迟队列存储的对象是对应的延迟消息，所谓的延迟消息是指当消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费

利用RabbitMq的TTL 和死信队列 来实现延时消费。

如果设置的是队列统一过期时间放到死信队列，没有什么问题。

如果是延时时间设置到每条消息上的。而不是给队列的。

实现方式为消息存活时间为动态用户页面可配置的。

这就导致了一个问题：

先用一条消息的存活时间是1天。后面又进了一条消息存活时间是1小时。

结果一小时到了，发现这条消息并没有被转发到消费延时过期消息的队列。

原因是尽管ttl是设给每条消息的。但是本质上，所有延时消息都还在一个队列里，对它过期时间的检测也是从头部开始的。

它不会检测每一条消息是否过期。而是顺序检测。

如果first in的消息过期时间很长，会导致它阻塞后进的消息。

不仅无法实现真正的过期时间。还会导致，一个大的过期时间的先进的消息，会堆积一堆后进的过期时间短的消息。

问题解决

这个时候可以使用rabbitMq的一个插件：rabbitmq_delayed_message_exchange

一段时间以来，人们一直在寻找用RabbitMQ实现延迟消息的传递方法，到目前为止，公认的解决方案是混合使用TTL和DLX。而rabbitmq_delayed_message_exchange插件就是基于此来实现的，RabbitMQ延迟消息插件新增了一种新的交换器类型，消息通过这种交换器路由就可以实现延迟发送

插件安装

需要根据自己的rabbitMq选择对应的版本。我rabbitMq的版本是RabbitMQ 3.11.0,对应的插件版本就是：3.11.1

基于Linux

--1、cd到rabbitmq默认安装位置
cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/plugins

--2、通过ftp工具将插件上传到此目录下

--3、开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

--4、重启MQ服务
systemctl restart rabbitmq-server

基于Docker

--1、通过ftp工具将插件上传到Linux服务器的根目录下

--2、拷贝到docker中rabbitmq插件目录下，rabbitmq_delayed_message_exchange-3.9.0.ez(下载包的全名)
docker cp /rabbitmq_delayed_message_exchange-3.9.0.ez 容器ID:/plugins

--3、进入容器
docker exec -it 容器id /bin/bash

--4、查看插件是否存在(确保2中的操作已经将插件拷贝过来了)
cd plugins
ls |grep delay

--5、开启插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

--6、退出容器
exit

--7、重启MQ服务
docker restart 容器ID

安装成功

web界面新建交换机选择类型出现红框标注即表示成功

代码实现

1：springBoot配置

@Configuration
public class DelayRabbitmqConfig {
 
    /**
     * 声明延迟队列
     * @return
     */
    @Bean
    public Queue delayQueue(){
        return new Queue(QueueConstant.DelayQueue,
                true,false,false);
    }
 
    /**
     * 声明延迟自定义交换机类型
     * @return
     */
    @Bean
    public CustomExchange delayCustomExchange(){
        HashMap<String, Object> args = new HashMap<>();
//        设置 x-delayed-type 为 direct，当然也可以是 topic 等 发送消息时设置消息头 headers 的 x-delay 属性，即延迟时间，如果不设置消息将会立即投递
        args.put("x-delayed-type","direct");
        return new CustomExchange(ExchangeConstant.DelayCustomerExchange,
                "x-delayed-message",true,false,args);
    }
 
    /**
     * 绑定延迟交换机和队列
     * @return
     */
    @Bean
    public Binding delayQueueAndCustomExchange(){
        return BindingBuilder.bind(delayQueue())
                .to(delayCustomExchange()).with(RoutingKeyConstant.DelayCustomerRoutingKey).noargs();
    }
}

springMvc配置

    引入依赖：
    xmlns:util="http://www.springframework.org/schema/util"
    http://www.springframework.org/schema/util
    http://www.springframework.org/schema/util/spring-util-4.0.xsd


	<!-- 定义延时队列 -->
	<!-- durable:是否持久化,宕机恢复后会重持久化日志恢复消息队列 -->
	<!-- exclusive: 仅创建者可以使用的私有队列，断开后自动删除 -->
	<!-- auto_delete: 当所有消费客户端连接断开后，是否自动删除队列 -->
	<rabbit:queue id="taskStartQueue" name="taskStartQueue" durable="true" auto-delete="false" exclusive="false"/>

    	<!--定义延时交换机类型-->
	<util:map id="delayExchangeArgs" map-class="java.util.HashMap">
		<entry key="x-delayed-type" value="direct"/>
	</util:map>

    <!-- 自定义交换机 - 延时交换机 ： rabbitMq 第三方插件 rabbitmq_delayed_message_exchange 实现延时队列 -->
    <!--注意：使用convertAndSend().发送消息后不会进入下面定义的队列中，而是由第三方插件管理，可以从交换机 messages delayed中查看消息数-->
    <bean id="taskExchange" name="delayExchange" class="org.springframework.amqp.core.CustomExchange">
        <constructor-arg name="name" value="taskExchange"/>
        <constructor-arg name="type" value="x-delayed-message"/>
        <constructor-arg name="durable" value="true"/>
        <constructor-arg name="autoDelete" value="false"/>
        <constructor-arg name="arguments" ref="delayExchangeArgs"/>
    </bean>

    <util:map id="emptyMap" map-class="java.util.HashMap" />

    <bean id="taskStartBind" class="org.springframework.amqp.core.Binding">
        <constructor-arg name="destination" value="taskStartQueue"/>
        <constructor-arg name="destinationType" value="QUEUE"/>
        <constructor-arg name="exchange" value="taskExchange"/>
        <constructor-arg name="routingKey" value="delay.task.start"/>
        <constructor-arg name="arguments" ref="emptyMap"/>
    </bean>

代码实现

//消息发送
final MessagePostProcessor messagePostProcessor = new MyMessagePostProcessor(Integer.valueOf(ttl.toString()));
DisTimingPushDto disTimingPushDto = new DisTimingPushDto();
disTimingPushDto.setOrderId(dispense.getOrderId());
disTimingPushDto.setPushTime(disDispense.getPushTime());
rabbitTemplate.convertAndSend(MsgQueueEnum.TIMING_PUSH.getExchangeName(), MsgQueueEnum.TIMING_PUSH.getQueueName(), disTimingPushDto, messagePostProcessor);

//每条消息时间配置
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

/**
 * 延迟消息处理器 Processor
 * @author king
 * @date 2022年12月28日 11:14
 */
public class MyMessagePostProcessor implements MessagePostProcessor {

    /**
     * 消息延迟时间，单位：毫秒
     */
    private final Integer TTL;

    public MyMessagePostProcessor(final Integer ttl) {
        this.TTL = ttl;
    }


    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setDelay(TTL);
        return message;
    }
}