死信队列TTL对队列进行延迟

我们知道可以通过TTL来对队列进行设置过期时间通过后置处理器MessagePostProcessor对消息进行设置过期时间

那么根据TTL及MessagePostProcessor机制可以处理关于延迟方面的问题。

比如秒杀之后给30分钟时间进行支付如果30分钟后没有支付订单取消。

比如餐厅订座,A用户早上8点预定的某家餐厅下午6点的座位,B用户中午12点预定的下午5点的座位根据场景我们需要的时先让B用户进行消费,然后A用户再消费;这时TTL和MessagePostProcessor延迟就已经不能满足订餐的场景了

因为TTL是对队列进行延迟,MessagePostProcessor是对消息进行延迟但是MessagePostProcessor对消息延迟是不能根据订座的时间去排序消费的

/***

比如当我们发送第一个消息时延迟的时间时50s,而发送的第二个消息延迟时间是30s,虽然延迟30s的消息比延迟50s发送的晚*

但按照我们设想的情况,延迟30s的消息应该先消费;可是实际情况却不是这样,而是延迟50s的消息到达时间后

场景:*

A用户和B用户预定餐厅,A用户先开始预定的,预定的是下午6点。

B用户比A用户预定操作晚一些,但是B用户预定的时间是下午5点。

通过此场景*

我们希望的是B用户先进行用餐(因为他预定的吃饭时间比A早一些,需要先安排吃饭。

不能说A用户没到6点B用户预定5点的吃不了。

),根据此*

之前的队列延迟还是消息延迟都不能满足场景需求了,这样就需要另一种延迟方式进行解决了!

注意*

TTL是对队列进行延迟,只要是在此队列中的消息都会按照TTL设置时间进行延迟;*

MessagePostProcessor是对消息进行延迟;**

如果我们不仅使用了消息延迟,而且还使用了队列延迟,那么延迟的时间就会以小的时间为准!*

理解:*

如果a消息设置的消息延迟是30s,b消息设置的延迟时间是90s,队列设置的延迟是60s。

那么a消息最终的延迟是30s(a的消息延迟与队列延迟*

比对以延迟时间小的为准),b消息最终延迟的时间是60s(b的消息延迟与队列延迟比对以延迟的时间小的为准)*/

1、死信队列TTL对队列进行延迟

给队列设置TTL过期时间此队列可不用绑定消费者,时间到后把消息投递到死信队列中由死信队列的消费者进行消费,这样就能达到延迟消费的作用

Beanpublic

QueueBuilder.durable(业务队列名称).deadLetterExchange(死信交换机名称).deadLetterRoutingKey(死信队列

消息停留时间//.maxLength(500).build();}监听死信队列即可处理超时的消息队列

上述实现方式中ttl延时队列中所有的消息超时时间都是一样的如果不同消息想设置不一样的超时时间就需要建立多个不同超时时间的消息队列比较麻烦且不利于维护。

对消息延迟

这样的方式可对每一个消息设置指定的过期时间,不用像TTL那样给队列设置过期时间(若队列设置的过期时间达到后其队列中的消息均会被删除或别的处理)但是由于队列是先进先出,若先投递的消息设置的过期时间是40s,后投递的消息过期时间是30s那么设置过期时间为30s的并不会到30s时就投递到死信队列中,而是等40s到期后才会一起被投递。

rabbitTemplate.convertAndSend(交换机名称,

RoutingKey,对象,

{message.getMessageProperties().setExpiration(String.valueOf(5000))//

RabbitMQ

message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return

message;});

该种方式可以创建一个承载不同超时时间消息的消息队列但是这种方式有一个问题如果消息队列中排在前面的消息没有到超时时间即使后面的消息到了超时时间先到超时时间的消息也不会进入死信队列而是先检查排在最前面的消息队列是否到了超时时间如果到了超时时间才会继续检查后面的消息。

使用rabbitmq的延时队列插件实现同一个队列中有多个不同超时时间的消息并按时间超时顺序出队

1、下载延迟插件

版本之后提供了一个插件rabbitmq-delayed-message-exchange来实现延迟队列

同时需保证

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases

2、安装插件并启用

/home/rabbitmq/rabbitmq_delayed_message_exchange-3.12.0.ez

rabbitmq:/plugins

rabbitmq_delayed_message_exchange

退出容器

xmlnshttp://maven.apache.org/POM/4.0.0

xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-parent/artifactIdversion2.7.1/version

!--

version2.2.5.RELEASE/version--relativePath/

!--

--/parentgroupIdcom.chensir/groupIdartifactIdspring-boot-rabbitmq/artifactIdversion0.0.1-SNAPSHOT/versionnamespring-boot-rabbitmq/namepropertiesjava.version8/java.versionhutool.version5.8.3/hutool.versionlombok.version1.18.24/lombok.version/propertiesdescriptionspring-boot-rabbitmq/descriptionpackagingpom/packagingmodulesmoduledirect-exchange/modulemodulefanout-exchange/modulemoduletopic-exchange/modulemodulegame-exchange/modulemoduledead-letter-queue/modulemoduledelay-queue/modulemoduledelay-queue2/module/modulesdependencyManagementdependenciesdependencygroupIdcn.hutool/groupIdartifactIdhutool-all/artifactIdversion${hutool.version}/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdversion${lombok.version}/version/dependency/dependencies/dependencyManagement/projectpom文件

?xml

xmlnshttp://maven.apache.org/POM/4.0.0

xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0

https://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersionparentgroupIdcom.chensir/groupIdartifactIdspring-boot-rabbitmq/artifactIdversion0.0.1-SNAPSHOT/versionrelativePath../pom.xml

/relativePath/parentartifactIddelay-queue2/artifactIddependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdcn.hutool/groupIdartifactIdhutool-all/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactIdoptionaltrue/optional/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-rabbit-test/artifactIdscopetest/scope/dependency/dependenciesbuildpluginsplugingroupIdorg.springframework.boot/groupIdartifactIdspring-boot-maven-plugin/artifactId/plugin/plugins/build/project配置文件

debug

spring.rabbitmq.host121.40.100.66

#默认5672

spring.rabbitmq.listener.simple.prefetch1

#表示消息确认方式其有三种配置方式分别是none、manual(手动)和auto(自动)默认auto

spring.rabbitmq.listener.simple.acknowledge-modeauto

spring.rabbitmq.listener.simple.retry.enabledtrue

#最大重试次数

spring.rabbitmq.listener.simple.retry.max-attempts5

#最大重试时间间隔

spring.rabbitmq.listener.simple.retry.max-interval3000ms

spring.rabbitmq.listener.simple.retry.initial-interval1000ms

spring.rabbitmq.listener.simple.retry.multiplier2

#决定被拒绝的消息是否重新入队默认是true与参数acknowledge-mode有关系

spring.rabbitmq.listener.simple.default-requeue-rejectedfalseconfig

package

org.springframework.amqp.core.*;

import

org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;

import

org.springframework.amqp.support.converter.MessageConverter;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;import

java.util.HashMap;

Jackson2JsonMessageConverter();}Beanpublic

CustomExchange

customExchange(){MapString,Object

args

HashMap();//延迟时以direct直连方式绑定args.put(x-delayed-type,direct);//

name:交换机名称

CustomExchange(chen-x-delayedExchange,x-delayed-message,true,false,args);}Beanpublic

Queue

QueueBuilder.durable(chen-DirectQueue).build();}Beanpublic

Binding

BindingBuilder.bind(directQueueLong()).to(customExchange()).with(direct123).noargs();}

生产者

org.springframework.amqp.rabbit.core.RabbitTemplate;

import

org.springframework.stereotype.Component;import

javax.annotation.Resource;Component

Slf4j

OrderIngOk();orderIngOk.setOrderNo(202207149687-1);orderIngOk.setId(1);orderIngOk.setUserName(倪海杉前-延迟40秒);rabbitTemplate.convertAndSend(chen-x-delayedExchange,

direct123,orderIngOk,m-{m.getMessageProperties().setDelay(40*1000);

//设置延迟时间对延迟交换机有效//

m.getMessageProperties().setExpiration(String.valueOf(30*1000));

m;});OrderIngOk

OrderIngOk();orderIngOk2.setOrderNo(202207149687-2);orderIngOk2.setId(2);orderIngOk2.setUserName(倪海杉后-延迟30秒);rabbitTemplate.convertAndSend(chen-x-delayedExchange,

direct123,orderIngOk2,m-{m.getMessageProperties().setDelay(30*1000);return

package

org.springframework.amqp.rabbit.annotation.RabbitHandler;

import

org.springframework.amqp.rabbit.annotation.RabbitListener;

import

org.springframework.stereotype.Component;import

Slf4j

{RabbitHandlerRabbitListener(queues

chen-DirectQueue

处理业务开始log.debug(接受到消息,并正常处理结束,{},

JSONUtil.toJsonStr(orderIngOk));//

处理业务结束}

com.chensir.provider.DirectProvider

消息生产完成

com.chensir.consumer.DirectConsumer

接受到消息,并正常处理结束,{id:2,OrderNo:202207149687-2,userName:倪海杉后-延迟30秒}

2023-08-29

com.chensir.consumer.DirectConsumer

接受到消息,并正常处理结束,{id:1,OrderNo:202207149687-1,userName:倪海杉前-延迟40秒}可见

延迟40s的是先发送的但是最终结果是先消费延迟30s的。

这样就能达到我们订座的场景需求了。