RabbitMQ死信机制实现延迟队列的实战

作者:wxd_1024 时间:2023-10-10 18:41:57 

延迟队列

延迟队列存储的对象肯定是对应的延时消息,所谓”延时消息”是指当消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费。

应用场景

三方支付,扫码支付调用上游的扫码接口,当扫码有效期过后去调用查询接口查询结果。实现方式:每当一笔扫码支付请求后,立即将此订单号放入延迟队列中(RabbitMQ),队列过期时间为二维码有效期,此队列没有设置消费者,过了有效期后消息会重新路由到指定的的队列,有消费者去执行订单查询。

RabbitMQ本身没有直接支持延迟队列功能,但是可以通过以下特性模拟出延迟队列的功能。 但是我们可以通过RabbitMQ的两个特性来曲线实现延迟队列:Time To Live(TTL)   和   Dead Letter Exchanges(DLX)

Time To Live(TTL)

RabbitMQ可以针对Queue和Message设置 x-message-tt,来控制消息的生存时间,如果超时,则消息变为dead letter(死信)RabbitMQ针对队列中的消息过期时间有两种方法可以设置。

A: 通过队列属性设置,队列中所有消息都有相同的过期时间。


<!-- 将消息放入此队列里,此队列设置过期时间,不制造消费者让其过期,过期后变成死信,消息会放入指定的新队列里,实现消息的延迟消费 -->
<rabbit:queue name="paycenter.scanpay.orderquery.delay.icbc" durable="true" auto-delete="false" exclusive="false" >
   <rabbit:queue-arguments>
       <entry key="x-message-ttl">
          <value type="java.lang.Long">${qrcode.expire.icbc}</value>
       </entry>
       <!--消息过期根据重新路由 -->
       <entry key="x-dead-letter-exchange" value="directExchange"/>
       <entry key="x-dead-letter-routing-key" value="paycenter.scanpay.orderquery"/>
   </rabbit:queue-arguments>
</rabbit:queue>

B: 对消息进行单独设置,每条消息TTL可以不同。


<!-- 将消息放入此队列里,次队列设置过期时间,不制造消费者让其过期,过期后变成死信,消息会放入指定的新队列里,实现消息的延迟消费 -->
<rabbit:queue name="paycenter.scanpay.orderquery.delay.icbc" durable="true" auto-delete="false" exclusive="false" >
   <rabbit:queue-arguments>
       <!--消息过期根据重新路由 -->
       <entry key="x-dead-letter-exchange" value="directExchange"/>
       <entry key="x-dead-letter-routing-key" value="paycenter.scanpay.orderquery"/>
   </rabbit:queue-arguments>
</rabbit:queue>

amqpTemplate.convertAndSend(mqMessage.getExchange(), mqMessage.getRoutingKey(), result, new ExpirationMessagePostProcessor(expireTime));

package com.emax.paycenter.mq.pruducer;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;

public class ExpirationMessagePostProcessor implements MessagePostProcessor {
private final Long ttl;

public ExpirationMessagePostProcessor(Long ttl) {
this.ttl = ttl;
}

@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setExpiration(ttl.toString());
return message;
}
}

如果同时使用,则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值,就成为dead letter

 Dead Letter Exchanges(DLX)

RabbitMQ的Queue可以配置x-dead-letter-exchange 和x-dead-letter-routing-key(可选)两个参数,如果队列内出现了dead letter,则按照这两个参数重新路由。
x-dead-letter-exchange:出现dead letter之后将dead letter重新发送到指定exchange
x-dead-letter-routing-key:指定routing-key发送
队列出现dead letter的情况有:
消息或者队列的TTL过期
队列达到最大长度
消息被消费端拒绝(basic.reject or basic.nack)并且requeue=false
利用DLX,当消息在一个队列中变成死信后,它能被重新publish到另一个Exchange。这时候消息就可以重新被消费。

RabbitMQ死信机制实现延迟队列的实战

注意一:ttl设置之后,下次修改时间,会报错,这时候,需要先删除该队列,重启项目。否则会报错。

注意二:消费者中,抛异常了没处理,会一直重复消费

注意三:下面的代码我模拟了1-10号消息,消息的内容里面是1-10。过期的时间是10-1秒。这里要注意,虽然10是第一个发送,但是它过期的时间最长。

过了10s以后,消费者开始收到数据,但是它是一次性收到如下结果:10、9 、8 、7 、6、5 、4 、3 、2 、1
Consumer第一个收到的还是10。10是第一个放进队列,但是它的过期时间最长。所以由此可见,即使一个消息比在同一队列中的其他消息提前过期,提前过期的也不会优先进入死信队列,它们还是按照入库的顺序让消费者消费。如果第一进去的消息过期时间是1小时,那么死信队列的消费者也许等1小时才能收到第一个消息。参考官方文档发现“Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered).”只有当过期的消息到了队列的顶端(队首),才会被真正的丢弃或者进入死信队列。

所以在考虑使用RabbitMQ来实现延迟任务队列的时候,需要确保业务上每个任务的延迟时间是一致的。如果遇到不同的任务类型需要不同的延时的话,需要为每一种不同延迟时间的消息建立单独的消息队列。(也可以考虑缓存队列,DelayQueue实现定时延迟执行任务,但是也有缺点:就是项目重启缓存里的数据就会丢失,DelayQueue的使用详见其他博文)


for(int i = 10; i>0; i-- ){
amqpTemplate.convertAndSend(mqMessage.getExchange(), mqMessage.getRoutingKey(), result, new ExpirationMessagePostProcessor(expireTime));
}

来源:https://blog.csdn.net/wxd_1024/article/details/83549967

标签:RabbitMQ,延迟队列
0
投稿

猜你喜欢

  • Android自定义View实现星星评分效果

    2023-10-27 23:10:42
  • java中main函数你知道多少

    2023-05-26 12:32:15
  • RN在Android打包发布App(详解)

    2021-08-25 06:21:11
  • C# 如何调用SAP RFC

    2023-07-02 18:11:42
  • Ubuntu中使用VS Code与安装C/C++插件的教程详解

    2023-07-26 08:21:00
  • C#二进制读写BinaryReader、BinaryWriter、BinaryFormatter

    2022-03-07 23:01:28
  • C#简单获取全屏中鼠标焦点位置坐标的方法示例

    2023-09-14 16:18:04
  • Java Web项目中解决中文乱码方法总结(三种最新方法)

    2023-07-18 18:13:22
  • C#同步网络时间的方法实例详解

    2021-10-24 15:47:47
  • 基于DateTime.ParseExact方法的使用详解

    2021-11-15 11:33:39
  • IDEA集成MyBatis Generator插件的使用

    2023-08-12 00:28:47
  • Android利用RecyclerView实现列表倒计时效果

    2023-09-04 05:29:27
  • 浅谈C#多线程简单例子讲解

    2023-06-24 03:03:50
  • c# 判断是否为空然后赋值的4种实现方法

    2021-06-06 22:01:12
  • Spring Boot集成Mybatis中如何显示日志的实现

    2023-04-09 10:49:48
  • SpringBoot配置 Druid 三种方式(包括纯配置文件配置)

    2021-06-03 01:41:19
  • fastjson转换对象实体@JsonProperty不生效问题及解决

    2023-10-07 00:13:51
  • java的Console类的使用方法及实例

    2022-09-20 15:47:27
  • 详解Spring缓存注解@Cacheable,@CachePut , @CacheEvict使用

    2021-11-18 12:18:05
  • C语言进制转换代码分享

    2021-10-16 15:39:23
  • asp之家 软件编程 m.aspxhome.com