RabbitMQ TTL机制实践建议

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId> <!-- 用于接口测试 -->
</dependency>

spring:
  rabbitmq:
    addresses: amqp://study:study@110.41.51.65:5672/bite 
    listener:
      simple:
        acknowledge-mode: manual # 手动确认模式，便于观察消息状态

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class TtlQueueConfig {
    // 常量：交换机、队列名称（参考文档命名）
    public static final String TTL_EXCHANGE_NAME = "ttl_exchange";
    public static final String TTL_QUEUE_WITH_TTL = "ttl_queue2"; // 带TTL的队列
    public static final String TTL_QUEUE_NO_TTL = "ttl_queue"; // 不带TTL的队列（用于对比）
    // 1. 声明Fanout交换机（广播模式，确保消息能被多个队列接收）
    @Bean("ttlExchange")
    public FanoutExchange ttlExchange() {
        return FanoutExchangeBuilder.fanoutExchange(TTL_EXCHANGE_NAME)
                .durable(true) // 持久化：服务重启后交换机不丢失
                .build();
    }
    // 2. 声明带TTL的队列（设置20秒过期）
    @Bean("ttlQueueWithTtl")
    public Queue ttlQueueWithTtl() {
        // 方式1：使用QueueBuilder.ttl()（文档推荐，简洁）
        return QueueBuilder.durable(TTL_QUEUE_WITH_TTL)
                .ttl(20 * 1000) // 20秒过期，单位：毫秒
                .build();
        // 方式2：使用withArguments() //底层方式
        // Map<String, Object> args = new HashMap<>();
        // args.put("x-message-ttl", 20000); // 20秒
        // return QueueBuilder.durable(TTL_QUEUE_WITH_TTL)
        //         .withArguments(args)
        //         .build();
    }
    // 3. 声明不带TTL的队列（用于对比过期效果）
    @Bean("ttlQueueNoTtl")
    public Queue ttlQueueNoTtl() {
        return QueueBuilder.durable(TTL_QUEUE_NO_TTL)
                .build();
    }
    // 4. 绑定：交换机与带TTL的队列
    @Bean("bindingWithTtl")
    public Binding bindingWithTtl(
            @Qualifier("ttlExchange") FanoutExchange exchange,
            @Qualifier("ttlQueueWithTtl") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }
    // 5. 绑定：交换机与不带TTL的队列
    @Bean("bindingNoTtl")
    public Binding bindingNoTtl(
            @Qualifier("ttlExchange") FanoutExchange exchange,
            @Qualifier("ttlQueueNoTtl") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }
}

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class TtlProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 发送消息到TTL交换机（同时投递到带TTL和不带TTL的队列）
    @RequestMapping("/ttl/queue")
    public String sendQueueTtlMessage() {
        String message = "Queue TTL test: " + System.currentTimeMillis();
        // Fanout交换机无需指定routingKey，设为空字符串
        rabbitTemplate.convertAndSend(TtlQueueConfig.TTL_EXCHANGE_NAME, "", message);
        return "消息发送成功（队列级TTL）：" + message;
    }
}

import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class TtlProducerController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 发送消息级TTL的消息（10秒过期）
    @RequestMapping("/ttl/message")
    public String sendMessageTtlMessage() {
        String message = "Message TTL test: " + System.currentTimeMillis();
        String ttlTime = "10000"; // 10秒过期，单位：毫秒（必须为字符串）
        // 通过MessagePostProcessor设置消息的expiration属性
        rabbitTemplate.convertAndSend(
                TtlQueueConfig.TTL_EXCHANGE_NAME, 
                "", 
                message,
                messagePostProcessor -> {
                    // 设置消息过期时间
                    messagePostProcessor.getMessageProperties().setExpiration(ttlTime);
                    return messagePostProcessor;
                }
        );
        return "消息发送成功（消息级TTL）：" + message;
    }
}

// 发送TTL=0的消息
@RequestMapping("/ttl/zero")
public String sendTtlZeroMessage() {
    String message = "TTL=0 test: " + System.currentTimeMillis();
    rabbitTemplate.convertAndSend(
            TtlQueueConfig.TTL_EXCHANGE_NAME,
            "",
            message,
            msgPostProcessor -> {
                msgPostProcessor.getMessageProperties().setExpiration("0"); // TTL=0
                return msgPostProcessor;
            }
    );
    return "TTL=0消息发送完成（无消费者则丢弃）";
}

// 为带TTL的队列绑定死信交换机
@Bean("ttlQueueWithDlx")
public Queue ttlQueueWithDlx() {
    return QueueBuilder.durable("ttl_queue_with_dlx")
            .ttl(20000) // 20秒过期
            .deadLetterExchange("dlx_exchange") // 绑定死信交换机
            .deadLetterRoutingKey("dlx.routing.key") // 死信路由键
            .build();
}