延時(shí)隊(duì)列我們可以簡(jiǎn)單粗暴的理解它為延時(shí)發(fā)送消息的隊(duì)列

好家伙，相當(dāng)粗暴

大致流程

配置類

@Configuration
public class DelayMQConfig {
    /**
     * 延時(shí)隊(duì)列交換機(jī)
     */
    public static final String DELAY_EXCHANGE_NAME = "delay.exchange";
    /**
     * 延時(shí)隊(duì)列
     */
    public static final String DELAY_QUEUE_NAME = "delay.queue";
    /**
     * 死信交換機(jī)
     */
    public static final String DEAD_EXCHANGE_NAME = "dead.exchange";
    /**
     * 死信隊(duì)列
     */
    public static final String DEAD_QUEUE_NAME = "dead.queue";

    /**
     * 聲明延時(shí)交換機(jī)
     */
    @Bean("delayExchange")
    public DirectExchange delayExchange(){
        return new DirectExchange(DELAY_EXCHANGE_NAME);
    }

    /**
     * 聲明延時(shí)隊(duì)列吠勘，并綁定到死信交換機(jī)上
     */
    @Bean("delayQueue")
    public Queue delayQueue(){
        Map<String, Object> args = new HashMap<>(2);
        //綁定死信交換機(jī)
        args.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
        //路由
        args.put("x-dead-letter-routing-key", "dead.routing.key");
        //延時(shí)10S
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();
    }

    /**
     * 延時(shí)隊(duì)列綁定到延時(shí)隊(duì)列交換機(jī)
     *
    **/
    @Bean
    public Binding delayBinding(@Qualifier("delayQueue")Queue queue,
                                @Qualifier("delayExchange")DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("delay.routing.key");
    }

    /**
     * 聲明死信交換機(jī)
     */
    @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");
    }
}

截圖看一下其中一個(gè)

延時(shí)隊(duì)列

消費(fèi)者

@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("收到消息的時(shí)間：{}，收到的消息內(nèi)容：{}", new Date().toString(), msg);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

生產(chǎn)者

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

發(fā)送一個(gè)消息峡眶，日志如下

2020-10-13 10:47:05.465  INFO 44044 --- [nio-9000-exec-1] com.lyy.study.controller.MyController    : 發(fā)送消息的時(shí)間：Tue Oct 13 10:47:05 CST 2020剧防，發(fā)送的消息內(nèi)容：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  : 收到消息的時(shí)間：Tue Oct 13 10:47:15 CST 2020，收到的消息內(nèi)容：delayMsgSend

到此一個(gè)簡(jiǎn)單的延時(shí)隊(duì)列就實(shí)現(xiàn)了辫樱。

(●'?'●)

不過更多的時(shí)候峭拘，我們其實(shí)需要的是給每一個(gè)消息添加自己的TTL，那么就會(huì)有一個(gè)問題拉，假如我們先發(fā)送了一個(gè)10S的延時(shí)消息鸡挠，又發(fā)送了一個(gè)5S的延時(shí)消息辉饱，預(yù)期的結(jié)果應(yīng)該是先消費(fèi)5S的那個(gè)消息，然而實(shí)際的結(jié)果是10S的消息先被消費(fèi)了拣展，才會(huì)消費(fèi)5S的那個(gè)消息（這個(gè)問題在TTL那篇文章有提過）彭沼，那怎么辦呢？

撓頭

rabbitmq_delayed_message_exchange插件

插件目錄