RabbitMQ中的死信队列与延迟队列

@Bean
public Queue bizQueue() {
    return QueueBuilder.durable("biz-queue")
            .deadLetterExchange("dlx-exchange")       // 死信交换机
            .deadLetterRoutingKey("dlx-routing-key")  // 死信路由键
            .ttl(10000)       // TTL 10秒
            .maxLength(5)     // 队列最大长度
            .build();
}

@Bean
public DirectExchange deadLetterExchange() {
    return ExchangeBuilder.directExchange("dlx-exchange").build();
}
@Bean
public Queue deadLetterQueue() {
    return QueueBuilder.durable("dlx-queue").build();
}
@Bean
public Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
    return BindingBuilder.bind(deadLetterQueue)
            .to(deadLetterExchange)
            .with("dlx-routing-key");
}

@RabbitListener(queues = "biz-queue", ackMode = "MANUAL")
public void onMessage(Message message, Channel channel) throws IOException {
    String body = new String(message.getBody());
    long deliveryTag = message.getMessageProperties().getDeliveryTag();
    if (body.contains("拒绝")) {
        // 拒绝消息，requeue=false → 进入死信队列
        channel.basicNack(deliveryTag, false, false);
    } else {
        channel.basicAck(deliveryTag, false);
    }
}

@RabbitListener(queues = "dlx-queue")
public void onDeadLetterMessage(Message message, Channel channel) throws IOException {
    String body = new String(message.getBody());
    System.out.println("[死信队列] 收到死信消息: " + body);
    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}

// 延迟队列（无消费者，20秒TTL后进入处理队列）
@Bean
public Queue delayQueue() {           
    return QueueBuilder.durable("delay-queue")
            .ttl(20000)                                          // 20秒TTL
            .deadLetterExchange("delay-process-exchange")        // 死信交换机
            .deadLetterRoutingKey("delay-process-routing-key")   // 死信路由键
            .build();
}
@Bean
public DirectExchange delayExchange() {    // 生产者第一次发送消息到达的交换机
    return ExchangeBuilder.directExchange("delay-exchange").build();
}
@Bean
public Binding delayBinding(Queue delayQueue, DirectExchange delayExchange) {       
    return BindingBuilder.bind(delayQueue).to(delayExchange).with("delay-routing-key");
}
// 处理队列（消费者监听这里）
@Bean
public Queue delayProcessQueue() {             //延迟后的消息放到这里迅速被消费者消费
    return QueueBuilder.durable("delay-process-queue").build();
}
@Bean
public DirectExchange delayProcessExchange() {     //延迟队列到达时间后消息进入这个交换机然后进入处理队列
    return ExchangeBuilder.directExchange("delay-process-exchange").build();
}
@Bean
public Binding delayProcessBinding(Queue delayProcessQueue, DirectExchange delayProcessExchange) {
    return BindingBuilder.bind(delayProcessQueue)
            .to(delayProcessExchange)
            .with("delay-process-routing-key");
}

// 生产者：发送消息到延迟队列
@GetMapping("/send")
public String send() {
    String now = LocalDateTime.now().format(FORMATTER);
    String message = "延迟消息，发送时间 " + now + "，预计20秒后被消费";
    rabbitTemplate.convertAndSend(
            DelayQueueConfig.DELAY_EXCHANGE,
            DelayQueueConfig.DELAY_ROUTING_KEY,
            message);
    return "消息已发送（" + now + "），约20秒后到达处理队列";
}
// 消费者：监听处理队列
@RabbitListener(queues = DelayQueueConfig.DELAY_PROCESS_QUEUE)
public void onMessage(String message) {
    String now = LocalDateTime.now().format(FORMATTER);
    System.out.println("[延迟队列消费者] 消费时间 " + now + "，消息内容: " + message);
}

#查看插件列表
rabbitmq-plugins list

#查看插件列表
rabbitmq-plugins list
#启动插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
#重启服务
service rabbitmq-server restart

@Bean
public DirectExchange pluginDelayExchange() {
    return ExchangeBuilder
            .directExchange("plugin-delay-exchange")
            .durable(true)
            .delayed()    // 关键：声明为延迟交换机
            .build();
}
@Bean
public Queue pluginDelayQueue() {
    return QueueBuilder.durable("plugin-delay-queue").build();
}
@Bean
public Binding pluginDelayBinding(Queue pluginDelayQueue, DirectExchange pluginDelayExchange) {
    return BindingBuilder.bind(pluginDelayQueue)
            .to(pluginDelayExchange)
            .with("plugin-delay-routing-key");
}

@GetMapping("/send")
public String send(@RequestParam(defaultValue = "5000") long delayMs) {
    String now = LocalDateTime.now().format(FORMATTER);
    String message = "插件延迟消息，发送时间 " + now + "，延迟 " + delayMs + "ms";
    rabbitTemplate.convertAndSend(
            PluginDelayQueueConfig.PLUGIN_DELAY_EXCHANGE,
            PluginDelayQueueConfig.PLUGIN_DELAY_ROUTING_KEY,
            message,
            msg -> {
                msg.getMessageProperties().setDelayLong(delayMs); // 每条消息独立延迟
                return msg;
            });
    return "消息已发送（" + now + "），约 " + delayMs + " 毫秒后被消费";
}

@RabbitListener(queues = PluginDelayQueueConfig.PLUGIN_DELAY_QUEUE)
public void onMessage(String message) {
    String now = LocalDateTime.now().format(FORMATTER);
    System.out.println("[插件延迟队列消费者] 消费时间 " + now + "，消息内容: " + message);
}