RabbitMQ 高级 - 死信队列

概述

DLX，全称为 Dead-Letter-Exchange，可以称之为死信交换机，也有人称之为死信邮箱。当消息在一个队列中变成死信（dead message）之后，他能被重新发送到另一个交换机中，这个交换机就是 DLX，绑定 DLX 的队列就称之为死信队列。消息变成死信，可能是由以下的原因

消息被拒绝
消息过期
队列达到最大长度

DLX 也是一个正常的交换机，和一般的交换机没哟区别，他能在任何的队列上被指定，实际上就是设置某一个队列的属性。当这个队列中存在死信时，RabbitMQ 就会自动将这个消息重新发布到设置的 DLX 上去，进而被路由到另一个队列，即死信队列。
想要使用死信队列，只需要在定义队列的时候设置队列参数 `x-dead-letter-exchange` 指定交换机即可。

页面中可以看到参数配置

死信队列配置

这就是一个普通的 direct 类型的消息队列

@Configuration
public class DeadRabbitMqConfiguration {
    @Bean
    public DirectExchange deadDirectExchange() {
        return new DirectExchange("dead_direct_exchange",true,false);
    }

    @Bean
    public Queue directDeadQueue() {
        return new Queue("dead.direct.queue",true);
    }

    @Bean
    public Binding deadBinding() {
        return BindingBuilder.bind(directDeadQueue()).to(deadDirectExchange()).with("dead");
    }

}

过期消息队列配置

相比于之前的消息队列，区别在于不仅指定过期时间，还指定了死信队列交换机以及死信队列名称。

@Configuration
public class TTLRabbitMqConfiguration {
    @Bean
    public DirectExchange ttlDirectExchange() {
        return new DirectExchange("ttl_direct_order_exchange",true,false);
    }

//    @Bean
//    public Queue directTtlQueue() {
//        // 设置过期时间
//        HashMap<String, Object> args = new HashMap<>();
//        args.put("x-message-ttl",5000); // 这里一定是一个int类型
//        return new Queue("ttl.direct.queue",true,false,false,args);
//    }

    @Bean
    public Queue directTtlQueue() {
        // 设置过期时间
        HashMap<String, Object> args = new HashMap<>();
        args.put("x-message-ttl",5000); // 这里一定是一个int类型
        args.put("x-dead-letter-exchange","dead_direct_exchange");
        args.put("x-dead-letter-routing-key","dead"); // 如果是 fanout 模式，是没有路由 key 的
        return new Queue("ttl2.direct.queue",true,false,false,args);
    }

    @Bean
    public Queue directTtlMessageQueue() {
        return new Queue("ttl.message.direct.queue",true);
    }

    @Bean
    public Binding ttlBinding() {
        return BindingBuilder.bind(directTtlQueue()).to(ttlDirectExchange()).with("ttl2");
    }


    @Bean
    public Binding ttlMessageBinding() {
        return BindingBuilder.bind(directTtlMessageQueue()).to(ttlDirectExchange()).with("ttlMessage");
    }

}

测试

/**
     * 队列设置过期时间
     * @param userId
     * @param productId
     * @param num
     */
    public void makeOrderTtl2(String userId,String productId,int num) {
        // 1:根据id查询商品是否充足
        // 2:保存订单
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单生成成功："+orderId);
        // 3:通过 MQ 来完成消息的分发
        // 交换机，路由 key/queue 队列名称，消息内容
        String exchangeName = "ttl_direct_order_exchange";
        String routingKey = "ttl2";
        rabbitTemplate.convertAndSend(exchangeName,routingKey,orderId);
    }

@Test
void testOrderTtl2() {
  orderService.makeOrderTtl2("1","1",12);
}

可以看到这个队列多了两个参数DLX和DLK，分别对应x-dead-letter-exchange和x-dead-letter-routing-key

过期的消息被投递到ttl2.direct.queue死信队列

本作品采用《CC 协议》，转载必须注明作者和本文链接