RabbitMQ死信队列

根本原理rabbitmq支持过期时间.
给队列设置过期时间15秒:那么进入此队列的所有消息都只会存活15秒
先看只给队列设置过期时间,其他不设置.

定义队列和交换机,注意给队列加上过期时间属性
过期时间不是指的队列过期,而是队列里的消息.

import java.util.HashMap;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue myQueue() {
        HashMap<String,Object> hashMap = new HashMap<>();
        hashMap.put("x-message-ttl", 15000);//定义队列里消息存活时间15秒
        return QueueBuilder.durable("myQueue").withArguments(hashMap).build();
        
    }
    @Bean
    public Exchange myExchange() {
        return new DirectExchange("myExchange");
    }
    @Bean
    public Binding myBinding(
            Queue myQueue,
            Exchange myExchange)
    {
        return BindingBuilder
                .bind(myQueue)
                .to(myExchange)
                .with("myKey")
                .noargs();
    }
}

写一个控制器,作为生产者,向队列推入消息

package com.spring_boot.controller;
import org.springframework.amqp.core.Message;
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;
import com.spring_boot.config.RabbitMQConfig;

@RestController
public class MQTestController { 
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @RequestMapping("deadMessage")
    public String deadMessage() {
        rabbitTemplate.convertAndSend("myExchange", "myKey", "这是消息");
        return "done";
    }
}

访问控制器,查看队列后台

图片.png

注意,我们并没有设置消费者,过了15秒,消息超时直接被队列丢弃了.
以下个人的理解,有错误的地方欢迎指出

到期被丢弃的消息就是死信.rabbitMQ有一种机制可以把死信投递到指定的交换机.再由交换机转发到合适的队列.
用来存放死信的队列就叫做死信队列.
用来把死信转发给队列的交换机叫做死信交换机

接下来尝试把过期的消息投递到死信队列

1. 首先应该有一个交换机来接收死信,还要有一个队列来保存死信
   在配置类中定义死信交换机和死信队列.(和普通的队列一样,叫法不同只是为了便于区分)

#RabbitConfig.java
/*
死信队列名为 deadQueue
死信交换机名为 deadExchange
死信路由键名为 deadKey
*/

    @Bean//死信队列
    public Queue deadQueue() {
        return new Queue("deadQueue");
    }
    @Bean//死信交换机
    public Exchange deadExchange() {
        return new DirectExchange("deadExchange");
    }
    @Bean//绑定
    public Binding deadBinding(Queue deadQueue,Exchange deadExchange) {
        return BindingBuilder
                .bind(deadQueue)
                .to(deadExchange)
                .with("deadKey") 
                .noargs();
    }

2. 我们必须告诉rabbitMQ把过期的消息投递给死信交换机
   回到最开始设置过期时间的地方,在队列定义时增加两个属性
   指明,死信应该以deadKey为路由键,投递到deadExchange交换机

    @Bean
    public Queue myQueue() {
        HashMap<String,Object> hashMap = new HashMap<>();
        hashMap.put("x-message-ttl", 15000);//定义消息存活时间15秒
        hashMap.put("x-dead-letter-exchange", "deadExchange");//设置此队列产生的死信会投递给哪个交换机
        hashMap.put("x-dead-letter-routing-key", "deadKey");//设置死信投递时使用的路由键
        return QueueBuilder.durable("myQueue").withArguments(hashMap).build();
    }

3. 手动在后台删掉之前生成的队列,和交换机重新运行项目,观察结果
   访问控制器, 生产者推送消息到myQueue 图片.png 由于myQueue设置了TTL=15000,15秒后,消息过期变成死信,被投递给死信交换机最终进入deadQueue. 图片.png

总结:
转换一下思维,商城的请在15分钟内支付应该就是这道理

1. 用户下单时,订单信息写入数据库,状态为待付款.
2. 订单ID放入myQueue,15分钟过期.
   (在这15分钟里,如果用户付款了,肯定会有回调函数把数据库里对应的支付状态改成已付款,订单ID继续留在队列里就好了)
3. 15分钟一到,订单ID就会进入死信队列
   那么只需要定义一个消费者,去消费死信队列里的消息.
   每取出一个订单ID就去数据库判断这个订单的支付状态,如果仍是未付款,就给他取消掉.支付成功的不用管

实验,给死信队列配置一个消费者

package com.spring_boot.consumer;

import java.text.SimpleDateFormat;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;



@Component
public class Receiver {
    @RabbitHandler
    @RabbitListener(queues = {"deadQueue"})
    public void cancelOrder(Message message) {
        //去数据库查询订单支付状态
        boolean paySuccess = false;
        if (!paySuccess) {
            System.out.println(new String(message.getBody()));
            System.out.println("订单超时,取消订单"+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(System.currentTimeMillis()));
        }   
    }
}

访问控制器,然后等待15秒,控制台输出 图片.png

)