RabbitMQ实现延时队列

什么是延迟队列？Producer将消息发送到消息队列服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息。例如超时关单、优惠券回收等场景会用到。

RabbitMQ本身是不支持延迟队列的，我们可以通过死信队列的特性来实现延时队列。

创建延时队列配置类

说明：创建死信队列和创建延时队列没什么不同，当成正常的队列创建就行了。

@Configuration
public class RabbitMQTTLConfig {

    /**
     * 死信队列
     */
    public static final String LOCK_MERCHANT_DEAD_QUEUE = "lock_merchant_dead_queue";

    /**
     * 死信交换机
     */
    public static final String LOCK_MERCHANT_DEAD_EXCHANGE = "lock_merchant_dead_exchange";

    /**
     * 进入死信队列的路由key
     */
    public static final String LOCK_MERCHANT_ROUTING_KEY = "lock_merchant_routing_key";


    /**
     * 创建死信交换机
     */
    @Bean("lockMerchantDeadExchange")
    public Exchange lockMerchantDeadExchange() {
        return new TopicExchange(LOCK_MERCHANT_DEAD_EXCHANGE, true, false);
    }

    /**
     * 创建死信队列
     */
    @Bean("lockMerchantDeadQueue")
    public Queue lockMerchantDeadQueue() {
        return QueueBuilder.durable(LOCK_MERCHANT_DEAD_QUEUE).build();
    }

    /**
     * 绑定死信交换机和死信队列，和创建普通队列没什么区别
     */
    @Bean
    public Binding lockMerchantBinding(@Qualifier("lockMerchantDeadQueue") Queue queue, @Qualifier("lockMerchantDeadExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(LOCK_MERCHANT_ROUTING_KEY).noargs();
    }
    
    
    

    /**
     * 普通队列，绑定的个死信交换机
     */
    public static final String NEW_MERCHANT_QUEUE = "new_merchant_queue";

    /**
     * 普通的topic交换机
     */
    public static final String NEW_MERCHANT_EXCHANGE = "new_merchant_exchange";

    /**
     * 路由key
     */
    public static final String NEW_MERCHANT_ROUTTING_KEY = "new_merchant_routing_key";


    /**
     * 创建普通交换机
     */
    @Bean("newMerchantExchange")
    public Exchange newMerchantExchange() {
        return new TopicExchange(NEW_MERCHANT_EXCHANGE, true, false);
    }

    /**
     * 创建普通队列
     */
    @Bean("newMerchantQueue")
    public Queue newMerchantQueue() {
        Map<String, Object> args = new HashMap<>(3);
        //消息过期后，进入到死信交换机
        args.put("x-dead-letter-exchange", LOCK_MERCHANT_DEAD_EXCHANGE);
        //消息过期后，进入到死信交换机的路由key
        args.put("x-dead-letter-routing-key", LOCK_MERCHANT_ROUTING_KEY);
        //过期时间，单位毫秒，10秒
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(NEW_MERCHANT_QUEUE).withArguments(args).build();
    }

    /**
     * 绑定交换机和队列
     */
    @Bean
    public Binding newMerchantBinding(@Qualifier("newMerchantQueue") Queue queue, @Qualifier("newMerchantExchange") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(NEW_MERCHANT_ROUTTING_KEY).noargs();
    }
}

延时队列-发送消息

思考一下，发送消息是发给谁的？发送消息是发给普通队列的，普通队列的消息过期会进入死信队列，然后我们监听死信队列的消息，但是发送消息是发给普通队列的。这次发消息不使用测试类发送了，换成发送

@RestController("/")
public class TestController {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send/ttl/{msg}")
    public boolean sendTTLMsg(@PathVariable String msg) {
        rabbitTemplate.convertAndSend(RabbitMQTTLConfig.NEW_MERCHANT_EXCHANGE, RabbitMQTTLConfig.NEW_MERCHANT_ROUTTING_KEY, 
                                      "超时关单信息，10秒后接收订单信息，" + msg);
        return true;
    }

}

延时队列-接收消息

消息消费者写法上没有不同，注意监听的是死信队列即可

@Component
@RabbitListener(queues = "lock_merchant_dead_queue") // 监听的队列名称是死信队列的名称
public class TTLMQListener {
    
    @RabbitHandler
    public void releaseCouponRecord(String msg, Message message) throws IOException {
        log.info("监听到延迟消息：消息内容，msg={}", msg);
    }
}