掌控消息全链路(3)——RabbitMQ/Spring-AMQP高级特性详解之TTL、死信和延迟

国内服务器5小时前发布 beixibaobao

1 0 0

🔥我的主页：
九转苍翎

⭐️个人专栏：
《Java SE 》
《Java集合框架系统精讲》
《MySQL高手之路：从基础到高阶》
《计算机网络》
《Java工程师核心能力体系构建》

天行健，君子以自强不息。

Java JDK版本：Oracle OpenJDK 17.0.9
SpringBoot版本：3.5.9

Spring Web
Lombok
Spring for RabbitMQ

RabbitMQ version：3.12.1
RabbitMQ实现延迟队列的插件：rabbitmq_delayed_message_exchange-3.12.0(已免费上传至我的资源)

1.TTL

TTL(Time-To-Live)是RabbitMQ中控制消息或队列生命周期的机制，用于在指定时间后自动删除消息或队列，避免资源堆积

消息TTL：为单条消息设置过期时间

队列TTL：设置整个队列的过期时间

声明和配置交换器、队列和绑定关系

import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean("ttlQueue")
    public Queue ttlQueue(){
        return QueueBuilder.durable(Constants.TTL_QUEUE).ttl(20000).build(); // 设置队列TTL为20秒
    }
    @Bean("ttlExchange")
    public DirectExchange ttlExchange(){
        return ExchangeBuilder.directExchange(Constants.TTL_EXCHANGE).build();
    }
    @Bean("ttlBinding")
    public Binding ttlBinding(@Qualifier("ttlExchange") DirectExchange directExchange,@Qualifier("ttlQueue") Queue queue){
        return  BindingBuilder.bind(queue).to(directExchange).with("ttl");
    }
}

发送消息

import jakarta.annotation.Resource;
import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Resource(name = "rabbitTemplate")
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/ttl")
    public String ttl(){
        MessagePostProcessor messagePostProcessor = message -> {
            message.getMessageProperties().setExpiration("10000"); // 设置消息TTL10秒
            return message;
        };
        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "TTL为10秒", messagePostProcessor);
        rabbitTemplate.convertAndSend(Constants.TTL_EXCHANGE, "ttl", "未设置TTL");
        return "发送成功";
    }
}

注意：两种TTL过期行为的触发条件不同

队列：超过TTL设置的时间后将该队列上的所有消息删除
消息：在即将投递时检查是否过期。为了避免为每条消息维护独立的计时器或对队列进行不间断的全量扫描带来的巨大性能开销，RabbitMQ采用了这种惰性检查策略
- 例如：一条已过期的消息位于队列中部，前面还有未过期的消息，它将不会被立即删除，而会滞留在队列中，直到它成为队列头部的下一条待投递消息时，才会被检查并丢弃

2.死信队列

死信(Dead Letter) 是指由于某些原因无法被正常消费的消息。RabbitMQ不会自动丢弃这些消息，而是可以将其路由到 死信交换器(Dead Letter Exchange，DLX) ，与DLX进行绑定的队列就是 死信队列(Dead Letter Queue，DLQ)

消息变成死信一般是由于以下几种情况：

消息被拒绝(Basic.Reject / Basic.Nack)，并且设置requeue为false

消息过期

队列达到最大长度

声明和配置交换器、队列和绑定关系

import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean("deadLetterQueue")
    public Queue deadLetterQueue(){
        return QueueBuilder.durable(Constants.DEAD_LETTER_QUEUE).build();
    }
    @Bean("deadLetterExchange")
    public DirectExchange deadLetterExchange(){
        return ExchangeBuilder.directExchange(Constants.DEAD_LETTER_EXCHANGE).build();
    }
    @Bean("deadLetterBinding")
    public Binding dlBinding(@Qualifier("deadLetterExchange") DirectExchange directExchange,@Qualifier("deadLetterQueue") Queue queue){
        return  BindingBuilder.bind(queue).to(directExchange).with("dead_letter");
    }
    @Bean("normalQueue")
    public Queue normalQueue(){
        return QueueBuilder.durable(Constants.NORMAL_QUEUE)
                .deadLetterExchange(Constants.DEAD_LETTER_EXCHANGE).deadLetterRoutingKey("dead_letter") // 绑定死信交换机
                .ttl(10000) // 测试过期消息
                .maxLength(10L) // 测试溢出的消息
                .build();
    }
    @Bean("normalExchange")
    public DirectExchange normalExchange(){
        return ExchangeBuilder.directExchange(Constants.NORMAL_EXCHANGE).build();
    }
    @Bean("normalBinding")
    public Binding normalBinding(@Qualifier("normalExchange") DirectExchange directExchange,@Qualifier("normalQueue") Queue queue){
        return  BindingBuilder.bind(queue).to(directExchange).with("normal");
    }
}

发送消息

import jakarta.annotation.Resource;
import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Resource(name = "rabbitTemplate")
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/deadLetter")
    public String deadLetter(){
        for (int i = 0; i < 20; i++) {
            rabbitTemplate.convertAndSend(Constants.NORMAL_EXCHANGE, "normal", "deadLetter" + i);
        }
        return "发送成功";
    }
}

配置消费者

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
@Component
@Slf4j
public class DeadLetterListener {
    @RabbitListener(queues = Constants.NORMAL_QUEUE)
    public void listenNormalQueue(Message message, Channel channel) throws Exception {
        log.info("normal_queue接收到消息:{},deliveryTag:{}",
                new String(message.getBody(), StandardCharsets.UTF_8),
                message.getMessageProperties().getDeliveryTag());
        try{
            int num = 3 / 0;
            log.info("处理成功");
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        }catch(Exception e){
            log.info("处理失败");
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), true,false);
        }
    }
    @RabbitListener(queues = Constants.DEAD_LETTER_QUEUE)
    public void listenDeadLetterQueue(Message message) {
        log.info("dead_letter_queue接收到消息:{},deliveryTag:{}",
                new String(message.getBody(), StandardCharsets.UTF_8),
                message.getMessageProperties().getDeliveryTag());
    }
}

3.延迟队列

延迟队列(Delayed Queue) 是一种特殊的消息队列，其中的消息不会立即被消费者获取，而是在指定的延迟时间过后才会变得可消费

用户下单后，订单信息通过fanout交换机转发到正常队列和ttlQueue

数据库消费正常队列并将订单信息写入数据库

数据库消费支付业务的队列，并完善订单的支付状态

30分钟后，消费者处理死信队列，判断支付状态。若已支付，不对数据库进行修改；若未支付，则取消或回滚该订单

3.1 ttl + DLX

RabbitMQ没有内置延迟队列功能，我们可以使用TTL+DLX组合的方式来实现延迟队列，如上述死信队列代码。但由于RabbitMQ对过期消息进行判断时，采用的是惰性检查策略，这会导致无法第一时间判断出消息是否过期

3.2 延迟插件

3.2.1 插件介绍

为了解决上述问题，RabbitMQ官方提供了⼀个延迟插件rabbitmq_delayed_message_exchange来实现延迟队列的功能

下载地址：rabbitmq_delayed_message_exchange

# 创建目录并将插件上传至该目录
root@VM-0-7-ubuntu:~# mkdir /usr/lib/rabbitmq/plugins
# 查看插件列表
root@VM-0-7-ubuntu:/usr/lib/rabbitmq/plugins# rabbitmq-plugins list

# 启动插件
root@VM-0-7-ubuntu:/usr/lib/rabbitmq/plugins# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 重启RabbitMQ服务
root@VM-0-7-ubuntu:/usr/lib/rabbitmq/plugins# service rabbitmq-server restart

3.2.1 代码编写

声明和配置交换器、队列和绑定关系

import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
    @Bean("delayQueue")
    public Queue delayQueue(){
        return QueueBuilder.durable(Constants.DELAY_QUEUE).build();
    }
    @Bean("delayExchange")
    public DirectExchange delayExchange(){
        return ExchangeBuilder.directExchange(Constants.DELAY_EXCHANGE).delayed().build();
    }
    @Bean("delayBinding")
    public Binding delayBinding(@Qualifier("delayExchange") DirectExchange directExchange,@Qualifier("delayQueue") Queue queue){
        return  BindingBuilder.bind(queue).to(directExchange).with("delay");
    }
}

发送消息

import jakarta.annotation.Resource;
import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class ProducerController {
    @Resource(name = "rabbitTemplate")
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/delay")
    public String delay(){
        MessagePostProcessor messagePostProcessor1 = message -> {
            message.getMessageProperties().setDelayLong(20000L);
            return message;
        };
        MessagePostProcessor messagePostProcessor2 = message -> {
            message.getMessageProperties().setDelayLong(10000L);
            return message;
        };
        rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "delay:20000", messagePostProcessor1);
        rabbitTemplate.convertAndSend(Constants.DELAY_EXCHANGE, "delay", "delay:10000", messagePostProcessor2);
        return "发送成功";
    }
}

配置消费者

import lombok.extern.slf4j.Slf4j;
import org.example.springrabbitmqextensions.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
@Component
@Slf4j
public class DelayListener {
    @RabbitListener(queues = Constants.DELAY_QUEUE)
    public void listenDelayQueue(Message message) {
        log.info("delay_queue接收到消息:{},deliveryTag:{}",
                new String(message.getBody(), StandardCharsets.UTF_8),
                message.getMessageProperties().getDeliveryTag());
    }
}