Springboot与RabbitMQ上手之TTL+DLX实现延迟队列（六）

前言

目的主要是学习RabbitMQ的TTL+DLX实现延迟队列，大概会简单介绍学习为主：毕竟还是要来演示Springboot整合RabbitMQ注解的方式来使用。

一. TTL

TTL，Time to Live的简称，即过期时间。RabbitMQ 可以对消息和队列设置TTL。
消息的 TTL 指的是消息的存活时间，RabbitMQ 支持消息、队列两种方式设置 TTL，分别如下：
消息设置 TTL：对消息的设置是在发送时进行 TTL 设置，通过 x-message-ttl 或expiration 字段设置，单位为毫秒，代表消息的过期时间，每条消息的 TTL 可不同。
队列设置 TTL：对队列的设置是在消息入队列时计算，通过 x-expires 设置，队列中的所有消息都有相同的过期时间，当超过了队列的超时设置，消息会自动的清除。
注意：如果以上两种方式都做了设置，消息的 TTL 则以两者之中最小的那个为准。

二.死信队列

DLX，全称为Dead-Letter-Exchange，可以称之为死信交换器，也有人称之为死信邮箱。当消息在一个队列中变成死信( dead message)之后，它能被重新被发送到另一个交换器中，这个交换器就是DLX，绑定DLX的队列就称之为死信队列。
消息变成死信一般是由于以下几种情况:
1). 消息被拒绝(Basic. Reject/Basic.Nack)，并且设置requeue参数为false;
2). 消息过期;
3). 队列达到最大长度。

2.1 DLX 交换机

DLX也是一个正常的交换器，和一般的交换器没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。当这个队列中存在死信时，RabbitMQ就会自动地将这个消息重新发布到设置的DLX上去，进而被路由到另一个队列，即死信队列。可以监听这个队列中的消息以进行相应的处理
####### 2.2 DLX声明
声明步骤如下：
第一步：声明正常的交换机，队列，路由键
第二步：在正常的队列里设置过期时间，绑定死信队列交换机名称以及设置发送给死信队列交换机路由键
第三步：声明死信交换机，队列，路由键
上面的做法是为了设置一个过期的正常队列路由到死信队列上面，让任意的队列去监听消费它。

image.png

    /* 设置队列过期时间*/
    @Bean
    public Exchange queueTimeOutExchange(){
        return new DirectExchange("queueTimeOut.exchange.test",true,false);
    }

    // 声明过期的队列并定义队列名称
    @Bean
    public Queue queueTimeOutQueue(){
        // 消息过期 特殊的args
        Map<String,Object> args  = new HashMap<>(16);
        args.put("x-message-ttl",20000);
        // 绑定死信交换机名称
        args.put("x-dead-letter-exchange", "dead.exchange.test");
        // 置发送给死信交换机的routingKey
        args.put("x-dead-letter-routing-key", "key.queue.dead");
        // 设置队列可以存储的最大消息数量
        args.put("x-max-length", 10);
        return new Queue("queueTimeOut.queue.test"
                ,true,false,false,args);
    }

    @Bean
    public Binding queueTimeOutBinding(){
        return new Binding("queueTimeOut.queue.test",
                Binding.DestinationType.QUEUE,
                "queueTimeOut.exchange.test",
                "key.queuetime.out",null);
    }

    /* 死信队列*/
    @Bean
    public Exchange deadExchange(){
        return new DirectExchange("dead.exchange.test",true,false);
    }

    // 声明过期的队列并定义队列名称
    @Bean
    public Queue deadQueue(){

        return new Queue("dead.queue.test"
                ,true,false,false);
    }

    @Bean
    public Binding deadBinding(){
        return new Binding("dead.queue.test",
                Binding.DestinationType.QUEUE,
                "dead.exchange.test",
                "key.queue.dead",null);
    }

image.png

上图标记的队列就是TTL+DLX。

上面标记之前为这个队列设置的特性

三，实现简单实例

3.1 生产端

service

    // 设置 死信队列
    public void deadMessag() throws JsonProcessingException;

impl

    /**
     * 測試死信隊列
     */
    @Override
    public void deadMessag() throws JsonProcessingException {
        /* 使用MessageProperties传递的对象转换成message*/
        MessageProperties messageProperties = new MessageProperties();
        OrderMessageDTO orderMessageDTO = new OrderMessageDTO();
        orderMessageDTO.setProductId(100);
        orderMessageDTO.setPrice(new BigDecimal("20"));
        orderMessageDTO.setOrderId(1);
        String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
        Message message = new Message(messageToSend.getBytes(),messageProperties);
        // 发送端确认是否确认消费
        CorrelationData correlationData = new CorrelationData();
        // 唯一ID
        correlationData.setId(orderMessageDTO.getOrderId().toString());
        rabbitTemplate.convertAndSend("dead.exchange.test","key.queue.dead",message,correlationData);
    }

controller

@RestController
@Slf4j
@RequestMapping("/api")
public class SendController {
    @GetMapping("/deadQueue")
    public void deadQueue() throws JsonProcessingException {
        directService.deadMessag();
    }
}

3.2 消费端