Springboot集成RabbitMQ死信队列的实现

更新时间：2021年09月16日 09:43:59 作者：小伙子你那什么车啊

在大多数的MQ中间件中，都有死信队列的概念。本文主要介绍了Springboot集成RabbitMQ死信队列的实现，具有一定的参考价值，感兴趣的小伙伴们可以参考一下

关于死信队列

在大多数的MQ中间件中，都有死信队列的概念。死信队列同其他的队列一样都是普通的队列。在RabbitMQ中并没有特定的“死信队列”类型，而是通过配置，将其实现。
当我们在创建一个业务的交换机和队列的时候，可以配置参数，指明另一个队列为当前队列的死信队列，在RabbitMQ中，死信队列（严格的说应该是死信交换机）被称为DLX Exchange。当消息“死掉”后，会被自动路由到DLX Exchange的queue中。

什么样的消息会进入死信队列？

1.消息的TTL过期。
2.消费者对broker应答Nack，并且消息禁止重回队列。
3.Queue队列长度已达上限。

场景分析

以用户订单支付为场景。在各大电商平台上，订单的都有待支付时间，通常为30min。当用户超过30min未支付订单，该订单的状态应该会变成“超时取消”，或类似的状态值的改变。
如果不使用MQ，可以设计一个定时任务，定时查询数据库，判断订单的状态和支付时间是否已经到期，若到期则修改订单的状态。但显然，这不是一个很好的操作，频繁访问数据库，造成不必要的资源浪费。
使用MQ，我们可以在下单的时候，当订单数据入库后，发送一条Message到Queue中，并设置过期时间为30min或自定义的支付过期时间。

   /**
     * 发送带有过期时间的消息
     */
    @GetMapping("/sendDlx")
    public void sendDlx() {
        Order order = new Order();
        order.setItemId(1);
        order.setStatus(1);
        rabbitTemplate.convertAndSend(orderExchange, orderRoutingKey, 
                JSON.toJSONString(order), message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            // 模拟，设置10S后消息过期
            message.getMessageProperties().setExpiration("10000");
            return message;
        });
    }

若30min后，还未有消费者（下游服务）消费这条消息，那么该条消息就会被路由到死信队列中。我们可以设置一个监听去监听死信队列，当收到死信队列的消息后，则根据消息数据，查询数据库订单状态是否还是待支付状态，若是，则修改成超时取消。

代码实现

以下是demo，未做服务的拆分，因此整个流程都是单个服务实现的，所以就没有下游服务，但并不影响整体业务。

RabbitMQConfig

将需要的交换机，队列，绑定都声明成SpringBean。Spring会自动创建这些到RabbitMQ服务中。
@Value注解部分都是配置文件exchange、queue、routingKey的名称。

/**
 * @author wulei
 */
@Configuration
public class RabbitConfig {

    @Value("${sunspring.order.exchange}")
    private String orderExchange;

    @Value("${sunspring.order.queue}")
    private String orderQueue;

    @Value("${sunspring.order.routingKey}")
    private String orderRoutingKey;

    @Value("${sunspring.dlx.exchange}")
    private String dlxExchange;

    @Value("${sunspring.dlx.queue}")
    private String dlxQueue;

    @Value("${sunspring.dlx.routingKey}")
    private String dlxRoutingKey;

    /**
     * 声明死信队列
     * @return DirectExchange
     */
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(dlxExchange);
    }

    /**
     * 声明死信队列
     * @return Queue
     */
    @Bean
    public Queue dlxQueue() {
        return new Queue(dlxQueue);
    }

    /**
     * 绑定死信队列到死信交换机
     * @return Binding
     */
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(dlxQueue())
                .to(dlxExchange())
                .with(dlxRoutingKey);
    }

    /**
     * 声明订单业务交换机
     * @return DirectExchange
     */
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(orderExchange);
    }

    /**
     * 声明订单业务队列
     * @return Queue
     */
    @Bean
    public Queue orderQueue() {
        Map<String,Object> arguments = new HashMap<>(2);
        // 绑定该队列到私信交换机
        arguments.put("x-dead-letter-exchange",dlxExchange);
        arguments.put("x-dead-letter-routing-key",dlxRoutingKey);
        return new Queue(orderQueue,true,false,false,arguments);
    }

    /**
     * 绑定订单队列到订单交换机
     * @return Binding
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
                .to(orderExchange())
                .with(orderRoutingKey);

    }
}

sunspring.order.exchange=sunspring_order_exchange
sunspring.order.queue=sunspring_order_queue
sunspring.order.routingKey=sunspring.order

sunspring.dlx.exchange=sunspring_dlx_exchange
sunspring.dlx.queue=sunspring.dlx.queue
sunspring.dlx.routingKey=dlx

在声明业务队列时，创建了一个Map，并且put了两个值，这两个值就是死信队列的声明。
x-dead-letter-exchange：死信交换机的名称
x-dead-letter-routing-key：死信交换机的路由键，因为demo中两个交换机的类型都是direct的，因此路由键必须相同。

/**
     * 声明订单业务队列
     * @return Queue
     */
    @Bean
    public Queue orderQueue() {
        Map<String,Object> arguments = new HashMap<>(2);
        // 绑定该队列到私信交换机
        arguments.put("x-dead-letter-exchange",dlxExchange);
        arguments.put("x-dead-letter-routing-key",dlxRoutingKey);
        return new Queue(orderQueue,true,false,false,arguments);
    }

监控页面

在exchange列表中有刚刚创建的业务交换机sunspring_order_exchange和死信交换机
sunspring_dlx_exchange

exchange列表

在Queue列表中，有死信队列sunspring_dlx_queue和业务队列sunspring_order_queue
并且业务队列上有DLX标记，可见当前队列已经绑定了一个死信队列。DLK表示的路由键。

queue列表

场景模拟

生产者

生产者发送了一个过期时间为10S的消息。
message.getMessageProperties().setExpiration(“10000”);

/**
     * 发送带有过期时间的消息
     */
    @GetMapping("/sendDlx")
    public void sendDlx() {
        Order order = new Order();
        order.setItemId(1);
        order.setStatus(1);
        rabbitTemplate.convertAndSend(orderExchange, orderRoutingKey,
                JSON.toJSONString(order), message -> {
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            message.getMessageProperties().setExpiration("10000");
            return message;
        });
    }

sunspring_order_queue接受到了一条消息，当前消息的状态是ready的，表示没有任何消费者消费这条消息。

在这里插入图片描述

10s后，当前消息路由到了死信队列中，sunspring_order_queue消息数量变成0，sunspring_dlx_queue数量变成1。

在这里插入图片描述

消费者，设置死信队列监听

通过设置对死信队列的监听，可以发现，在Springboot启动之后，创建了对RabbitMQ的监听，死信队列的消息也立刻被消费了。

因此，我们可以监听死信队列，对未被消费的消息进行下一步操作。如场景分析中的更改订单状态。

   @RabbitListener(queues = "sunspring.dlx.queue")
    public void dlxListener(Message message,Channel channel) throws IOException {
        System.out.println(new String(message.getBody()));

        //对消息进行业务处理....
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }

2019-08-20 20:05:05.158 INFO 4420 --- [ main] o.s.a.r.c.CachingConnectionFactory : Attempting to connect to: [120.27.243.91:5672]
2019-08-20 20:05:05.224 INFO 4420 --- [ main] o.s.a.r.c.CachingConnectionFactory : Created new connection: rabbitConnectionFactory#68ab0936:0/SimpleConnection@74606204 [delegate=amqp://guest@120.27.243.91:5672/, localPort= 13563]
{"itemId":1,"status":1}

到此这篇关于Springboot集成RabbitMQ死信队列的实现的文章就介绍到这了,更多相关Springboot RabbitMQ死信队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

knife4j3.0.3整合gateway和注册中心的详细过程
这篇文章主要介绍了knife4j3.0.3整合gateway和注册中心的详细过程，本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
2023-03-03
如何解决springboot自动重启问题
这篇文章主要介绍了如何解决springboot自动重启问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2023-09-09
浅谈Java的WeakHashMap源码
这篇文章主要介绍了浅谈Java的WeakHashMap源码,WeakHashMap,从名字可以看出它是某种 Map,它的特殊之处在于 WeakHashMap 里的entry可能会被GC自动删除,即使程序员没有调用remove()或者clear()方法,需要的朋友可以参考下
2023-09-09
idea 列编辑模式取消的操作
这篇文章主要介绍了idea 列编辑模式取消的操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2021-02-02
Spring Cache与Redis结合的使用方式
这篇文章主要介绍了Spring Cache与Redis结合的使用方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-12-12
关于Spring 中 StringUtils.isEmpty 被弃用如何正确使用
SpringBoot/SpringFramework升级后,StringUtils.isEmpty()方法已被弃用,推荐使用hasLength和hasText替代,hasLength判断是否有长度,hasText判断是否包含至少一个非空白字符,本文给大家介绍关于Spring 中 StringUtils.isEmpty 被弃用如何正确使用,感兴趣的朋友一起看看吧
2026-01-01
Java HashMap两种简便排序方法解析
这篇文章主要介绍了Java HashMap两种简便排序方法解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-01-01
SpringBoot自动配置深入探究实现原理
在springboot的启动类中可以看到@SpringBootApplication注解，它是SpringBoot的核心注解，也是一个组合注解。其中@SpringBootConfiguration、@EnableAutoConfiguration、@ComponentScan三个注解尤为重要。今天我们就来浅析这三个注解的含义
2022-08-08
Java使用条件语句和循环结构确定控制流(实例)
下面小编就为大家带来一篇Java使用条件语句和循环结构确定控制流(实例)。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-06-06
Java并发编程之原子性-Atomic的使用
这篇文章主要介绍了Java并发编程之原子性-Atomic的使用方式，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-03-03