rabbitMQ-延时队列

延时队列我们可以简单粗暴的理解它为延时发送消息的队列

好家伙，相当粗暴

那延时队列的应用场景有哪些呢，比如订单在一段时间内未支付则取消订单，就是需要在某个事件发生之后或者之前的某个时间点完成另一事件。
这种场景讲道理可以用定时器来完成，但是如果有些事件的时间点需要精确到秒，我们就需要每秒轮询一次，或者在数据量较大的时候，定时器可能需要跑很久，给系统带来很大压力！（好了就这把，编不下去了。。QAQ
那rabbitMQ的TTL只是消息在队列中的存活时间，并不是说消息在存活一段时间后才会发送给消费者，所以并不能只用TTL来实现延时队列，还需要借助死信队列。

大致流程

生产者发送一条延时消息到延时队列，当消息过期又会被路由到死信队列，消费者只要监听死信队列即可。

配置类

@Configuration
public class DelayMQConfig {
    /**
     * 延时队列交换机
     */
    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
    /**
     * 延时队列
     */
    public static final String DELAY_QUEUE_NAME = "delay.queue";
    /**
     * 死信交换机
     */
    public static final String DEAD_EXCHANGE_NAME = "dead.exchange";
    /**
     * 死信队列
     */
    public static final String DEAD_QUEUE_NAME = "dead.queue";

    /**
     * 声明延时交换机
     */
    @Bean("delayExchange")
    public DirectExchange delayExchange(){
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }

    /**
     * 声明延时队列，并绑定到死信交换机上
     */
    @Bean("delayQueue")
    public Queue delayQueue(){
        Map<String, Object> args = new HashMap<>(2);
        //绑定死信交换机
        args.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
        //路由
        args.put("x-dead-letter-routing-key", "dead.routing.key");
        //延时10S
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();
    }

    /**
     * 延时队列绑定到延时队列交换机
     *
    **/
    @Bean
    public Binding delayBinding(@Qualifier("delayQueue")Queue queue,
                                @Qualifier("delayExchange")DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("delay.routing.key");
    }

    /**
     * 声明死信交换机
     */
    @Bean("deadExchange")
    public DirectExchange deadExchange(){
        return new DirectExchange(DEAD_EXCHANGE_NAME);
    }

    @Bean("deadQueue")
    public Queue deadQueue(){
        return new Queue(DEAD_QUEUE_NAME);
    }

    @Bean
    public Binding deadBinding(@Qualifier("deadQueue")Queue queue,
                               @Qualifier("deadExchange")DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("dead.routing.key");
    }
}

截图看一下其中一个

延时队列

消费者

@Component
public class DelayListener {
    private Logger logger = LoggerFactory.getLogger(getClass());
    @RabbitListener(queues = DelayMQConfig.DEAD_QUEUE_NAME)
    public void receive(Message message, Channel channel) throws Exception{
        String msg = new String(message.getBody());
        logger.info("收到消息的时间：{}，收到的消息内容：{}", new Date().toString(), msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

生产者

    @PostMapping("delayMsgSend")
    public void delayMsgSend(@RequestParam("msg")String msg){
        rabbitTemplate.convertSendAndReceive(DelayMQConfig.DELAY_EXCHANGE_NAME,"delay.routing.key",msg);
    }

发送一个消息，日志如下

2020-10-13 10:47:05.465  INFO 44044 --- [nio-9000-exec-1] com.lyy.study.controller.MyController    : 发送消息的时间：Tue Oct 13 10:47:05 CST 2020，发送的消息内容：delayMsgSend
2020-10-13 10:47:05.532  INFO 44044 --- [nio-9000-exec-1] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService
2020-10-13 10:47:05.545  INFO 44044 --- [nio-9000-exec-1] .l.DirectReplyToMessageListenerContainer : Container initialized for queues: [amq.rabbitmq.reply-to]
2020-10-13 10:47:05.627  INFO 44044 --- [nio-9000-exec-1] .l.DirectReplyToMessageListenerContainer : SimpleConsumer [queue=amq.rabbitmq.reply-to, consumerTag=amq.ctag-KMeIYI6Pcun_gzAm5AYe3w identity=49703e41] started
2020-10-13 10:47:15.675  INFO 44044 --- [ntContainer#0-1] com.lyy.study.mq.listener.DelayListener  : 收到消息的时间：Tue Oct 13 10:47:15 CST 2020，收到的消息内容：delayMsgSend

到此一个简单的延时队列就实现了。

(●'◡'●)

不过更多的时候，我们其实需要的是给每一个消息添加自己的TTL，那么就会有一个问题拉，假如我们先发送了一个10S的延时消息，又发送了一个5S的延时消息，预期的结果应该是先消费5S的那个消息，然而实际的结果是10S的消息先被消费了，才会消费5S的那个消息（这个问题在TTL那篇文章有提过），那怎么办呢？

挠头

不要慌，https://www.rabbitmq.com/community-plugins.html从官网上下载一个叫rabbitmq_delayed_message_exchange的插件就好了！

rabbitmq_delayed_message_exchange插件

下载之后解压放到rabbitMQ的插件目录

插件目录

然后进入sbin目录，执行以下命令
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
重启就OK了！
验证的部分我就不写了，懒惰.jpg