RabbitMQ/Spring-AMQP 高级特性：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");
    }
}

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

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

2. 死信队列

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

1. 消息被拒绝 (Basic.Reject / Basic.Nack)，并且设置 requeue 为 false
2. 消息过期
3. 队列达到最大长度

配置消费者

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());
    }
}

发送消息

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 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");
    }
}

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.2 代码编写

配置消费者

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());
    }
}

发送消息

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 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");
    }
}