RabbitMQ养成记（10.高级特性：死信队列，延迟队列）

news2024/11/25 12:40:53

死信队列（DLX）

这个概念在其他MQ产品里面也是有的，只不过在Rabbitmq中稍微特殊一点
什么叫私信队列呢？就是当消息成为 dead message之后，可以重新发到另外一台交换机，这个交换机就是DLX。

在这里插入图片描述

注意这里的有翻译歧义，这里的DLX 指的是交换机，而不是一个队列。

消息成为死信队列的三种情况

队列的消息长度到达限制。
消费者拒收消息，

            channel.basicNack(deliveryTag,true,false);
            
            //这里拒收消息
            /**
             * deliveryTag 标识
             * requeue 是否打回原队列如果为false 则进入死信队列
             */
            channel.basicReject(deliveryTag,false);
        }

存在消息过期设置消息超时未消费（就是上一篇中的TTL）

我们拿这个TTL举例：

首先你在发送端需要有2个队列一个正常队列一个私信队列
然后你在创建延时队列的时候要注明这个队列的消息一旦超时那么送到哪个死信交换机的队列里面去：

@Configuration
public class RabbitMQConfig {
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    public static final String QUEUE_NAME_TTL = "boot_queue_ttl";

    public static final String DEAD_EXCHANGE_NAME = "dead_topic_exchange";
    public static final String DEAD_QUEUE_NAME = "dead_queue";

    //正常交换机

    @Bean("bootExchange")
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }


    @Bean("bootQueue")
    public Queue bootQueue(){
        Map<String, Object> map = new HashMap<>();
        // 设置TTL
        map.put("x-message-ttl", 20000);
        // 设置死信的目的交换机
        map.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
        // 设置死信交给目的交换机时的路由键
        map.put("x-dead-letter-routing-key", "boot.111");
        return QueueBuilder.durable(QUEUE_NAME_TTL).withArguments(map).build();
    }


    @Bean
    public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchange") Exchange exchange){
       return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }

    //死信交换机
    @Bean("deadExchange")
    public Exchange deadExchange(){
        return ExchangeBuilder.topicExchange(DEAD_EXCHANGE_NAME).durable(true).build();
    }

    @Bean("deadQueue")
    public Queue deadQueue(){
        return QueueBuilder.durable(DEAD_QUEUE_NAME).build();
    }

    @Bean
    public Binding bindDeadQueueExchange(@Qualifier("deadQueue") Queue queue,@Qualifier("deadExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("boot.*").noargs();
    }


}

好了我们发送消息往正常队列里面发：
在这里插入图片描述

看超时之后它进来了：
在这里插入图片描述

·这是超时的消息进入DLX的情况
再动动手我们实践一下消费者拒收之后进入死信的情况：

我们监听正常的队列然后接受到消息之后
我们故意写一个bug 触发exception
调用basicNack方法打回这个消息让它进入dlx
这里千万要注意：basicNack第三个参数要为false 才会进入dlx

如果为true 那它就会打回到正常队列然后不停的给你发

    @RabbitListener(queues = "boot_queue_ttl")
    public void ListenerQueue(Message message,
                              @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
                              Channel channel) throws IOException {
        try {
            System.out.println("收到消息为"+new String(message.getBody()));

            System.out.println("处理业务逻辑。。");

//            int i =3/0; //手动制造一个bug 让程序进入异常

            channel.basicAck(deliveryTag,true);

        }catch (Exception e){
            //拒绝签收
            //这里的第三个参数表示 如果为true 消息重回到queue broker会重新发送该消息给消费端
            System.out.println(" error 处理业务逻辑失败 消息打回 进入私信队列");
            channel.basicNack(deliveryTag,true,false);
}