过期时间 TTL

1、概述

过期时间TTL指可以对消息设置预期的时间，在这个时间内都可以被消费者接收获取；过后将被自动删除，RabbitMQ设置TTL的两种方式：

• 通过队列属性设置，队列中所有的消息都有相同的过期时间
• 对消息进行单独的设置，每条消息的TTL不同

如果同时使用以上两种方式，则以较小的那个时间为准。消息在队列的生存时间一旦超过设置的TTL值，将被投递到死信队列，成为dead message，消费者将无法再收到该消息。

过期队列：消息存入死信队列
消息设置过期时间：过期后，消息直接移除

创建队列

@Configuration
public class TTLRabbitMQConfig {

    //1、声明注册direct模式的交换机
    @Bean
    public DirectExchange ttlDirectExchange() {
        return new DirectExchange("ttl_direct_exchange", true, false);
    }

    //2、声明队列过期时间
    @Bean
    public Queue directTTLQueue() {
        Map<String, Object> args = new HashMap<>();
        //单位 ms
        args.put("x-message-ttl", 5000);
        return new Queue("ttl.direct.queue", true, false, false, args);
    }

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

    //过期队列
    @Bean
    public Binding ttlQueue(){
        return BindingBuilder.bind(directTTLQueue()).to(ttlDirectExchange()).with("ttl");
    }

    //消息含有过期时间，队列没有设置过期时间
    @Bean
    public Binding ttlMessageQueue(){
        return BindingBuilder.bind(directTTLMessageQueue()).to(ttlDirectExchange()).with("ttlMessage");
    }
    
}

生产者

@Service
public class OrderService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    // ttl
    public void makeOrderTtl(String userId, String productId, int num) {
        //1、根据商品id查询库存
        //2、保存订单
        String orderId = UUID.randomUUID().toString().replaceAll("-", "");
        System.out.println("创建订单成功：" + orderId);
        //3、通过 MQ 来完成消息的分发
        String exchangeName = "ttl_direct_exchange";
        String routingKry = "ttl";
        //交换机   路由key/queue队列名  消息内容
        rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId);
    }

    // ttl 单独设置过期时间
    public void makeOrderTtlMessage(String userId, String productId, int num) {
        //1、根据商品id查询库存
        //2、保存订单
        String orderId = UUID.randomUUID().toString().replaceAll("-", "");
        System.out.println("创建订单成功：" + orderId);
        //3、通过 MQ 来完成消息的分发
        String exchangeName = "ttl_direct_exchange";
        String routingKry = "ttlMessage";

        //设置消息过期时间
        MessagePostProcessor message = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setExpiration("5000");
                message.getMessageProperties().setContentEncoding("UTF-8");
                return message;
            }
        };

        //交换机   路由key/queue队列名  消息内容
        rabbitTemplate.convertAndSend(exchangeName, routingKry, orderId, message);
    }
}

消费者

//消费者代码同理fanout模式的消费者