免费的素材库百度seo搜搜

1.概述

延迟队列其实就是队列里的消息是希望在指定时间到了以后或之前取出和处理，简单来说，延时队列就是用来存放需要在指定时间被处理的元素的队列。

延时队列的使用场景：

1.订单在十分钟之内未支付则自动取消

2.新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。

3.用户注册成功后，如果三天内没有登陆则进行短信提醒。

4.用户发起退款，如果三天内没有得到处理则通知相关运营人员。

5.预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议

2.代码演示 

代码是用springboot整合的。

先导入依赖

<dependencies><!--RabbitMQ 依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.47</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><!--RabbitMQ 测试依赖--><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency>
</dependencies>

配置文件

spring.rabbitmq.host=192.168.10.137
spring.rabbitmq.port=5672
spring.rabbitmq.username=rabbit
spring.rabbitmq.password=rabbit
spring.rabbitmq.virtual-host=/

启动器

@SpringBootApplicationpublic class App {public static void main(String[] args) {SpringApplication.run(App.class,args);}}

 需求如下：创建两个队列 QA 和 QB，两者队列 TTL 分别设置为 10S 和 40S，然后在创建一个交 换机 X 和死信交 换机 Y，它们的类型都是 direct，创建一个死信队列 QD，它们的绑定关系如下：

定义配置类，描述上图的队列，交换机以及队列和交换机之间的关系

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;// 使用@Configuration注解表明这是一个配置类，Spring容器会扫描该类来获取Bean的定义信息
@Configuration
public class DelayedQueueConfig {// 定义直连类型（Direct）的交换机，名称为xExchange// @Bean注解用于将方法返回的对象注册为Spring容器中的一个Bean，"xExchange"是该Bean的名称@Bean("xExchange")public DirectExchange xExchange() {// 创建并返回一个名为"X"的DirectExchange实例，DirectExchange类型的交换机根据路由键直接转发消息return new DirectExchange("X");}// 声明另一个直连类型的交换机，名称为yExchange@Bean("yExchange")public DirectExchange yExchange() {// 创建并返回一个名为"Y"的DirectExchange实例return new DirectExchange("Y");}// 声明队列queueA@Bean("queueA")public Queue queueA() {// 创建一个HashMap用于存储队列的属性参数Map<String, Object> args = new HashMap<>();// 设置死信交换机（当消息在队列中过期或被否定确认等情况时，消息会被转发到这个交换机）为yExchangeargs.put("x-dead-letter-exchange", "Y");// 设置死信路由键，当消息进入死信交换机后，根据这个路由键来路由到对应的死信队列args.put("x-dead-letter-routing-key", "YD");// 设置消息在队列中的存活时间（即延时时间）为10000毫秒，也就是10秒，这使得queueA成为一个延时队列args.put("x-message-ttl", 10000);// 创建一个持久化的队列（QueueBuilder.durable方法），名称为"QA"，并带上前面设置的属性参数return QueueBuilder.durable("QA").withArguments(args).build();}// 绑定交换机xExchange和队列queueA// @Qualifier注解用于根据指定的Bean名称来注入对应的Bean实例，这里分别指定了要注入的队列和交换机实例@Beanpublic Binding queueQABindingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange) {// 使用BindingBuilder将队列queueA绑定到交换机xExchange上，绑定的路由键为"XA"return BindingBuilder.bind(queueA).to(xExchange).with("XA");}// 声明队列queueB，设置其延时时间为40秒@Bean("queueB")public Queue queueB() {Map<String, Object> args = new HashMap<>();// 同样设置死信交换机为yExchangeargs.put("x-dead-letter-exchange", "Y");// 设置死信路由键为"YD"args.put("x-dead-letter-routing-key", "YD");// 设置消息在队列中的存活时间为40000毫秒，即40秒，使queueB成为延时队列args.put("x-message-ttl", 40000);// 创建一个持久化的队列，名称为"QB"，并带上相关属性参数return QueueBuilder.durable("QB").withArguments(args).build();}// 绑定交换机xExchange和队列queueB@Beanpublic Binding queueQBBindingX(@Qualifier("queueB") Queue queueB,@Qualifier("xExchange") DirectExchange xExchange) {// 将队列queueB绑定到交换机xExchange上，绑定的路由键为"XB"return BindingBuilder.bind(queueB).to(xExchange).with("XB");}// 声明死信队列queueD@Bean("queueD")public Queue queueD() {// 创建一个名称为"QD"的队列，用于接收从延时队列中过期转移过来的消息return new Queue("QD");}// 声明死信交换机yExchange和死信队列queueD的绑定关系@Beanpublic Binding deadQueueBindingQD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange) {// 使用BindingBuilder将死信队列queueD绑定到死信交换机yExchange上，绑定的路由键为"YD"return BindingBuilder.bind(queueD).to(yExchange).with("YD");}
}

 生产者

// @RestController注解表明该类是一个RESTful风格的控制器，用于处理HTTP请求并返回JSON等格式的数据
@RestController 
// @RequestMapping("ttl")注解用于映射请求路径，所有以"/ttl"开头的请求会被该控制器处理
@RequestMapping("ttl") 
public class SendMessageController { // @Resource注解用于自动装配RabbitTemplate实例，RabbitTemplate是Spring AMQP提供的用于操作RabbitMQ的工具类@Resource RabbitTemplate rabbitTemplate; // 该方法用于处理"/sendMsg/{message}"路径的请求，是消息发送的逻辑所在// @RequestMapping("sendMsg/{message}")注解将该方法映射到指定的请求路径，其中{message}是一个路径变量@RequestMapping("sendMsg/{message}") public void sendMessage(@PathVariable("message") String message){ // 使用rabbitTemplate的convertAndSend方法向RabbitMQ发送消息// 第一个参数"X"指定交换机名称，对应前面配置的xExchange// 第二个参数"XA"指定路由键，用于将消息路由到绑定了该路由键的队列（这里是queueA）// 消息内容是拼接后的字符串，表明消息来自ttl为10秒钟的延时队列，并带上传入的参数messagerabbitTemplate.convertAndSend("X", "XA", "消息来自ttl为10秒钟的延时队列" + message); // 同理，这条消息发送到交换机"X"，通过路由键"XB"路由到queueB// 消息内容表明来自ttl为40秒钟的延时队列rabbitTemplate.convertAndSend("X", "XB", "消息来自ttl为40秒钟的延时队列" + message); }
}

消费者

// @Component注解将该类标记为一个Spring组件，使其能被Spring容器扫描并管理
@Component 
// @Slf4j注解是Lombok提供的，用于自动生成日志对象log，方便在类中记录日志
@Slf4j 
public class MessageConsumerListener { // @RabbitListener(queues = "QD")注解表明该方法是一个RabbitMQ消息监听器，监听名为"QD"的队列// 当队列"QD"（即前面配置的死信队列）中有消息时，该方法会被触发执行@RabbitListener(queues = "QD") public void getMessage(Message message, Channel channel) throws Exception{ // 获取消息体内容，将消息的字节数组转换为字符串String msg = new String(message.getBody()); // 使用日志对象log记录信息，输出当前时间以及从死信队列收到的消息内容log.info("当前时间是:{},收到死信队列的消息{}",new Date().toString(),msg); }
}

我们上面构建的延时队列太局限性了，因为我们直接写死了延时队列的时间，但我们实际的应用中很多情况都是根据客户端动态设置时间，比如腾讯会议我们要预定多久的会。

所以下面这个案例新增了一个队列QC，他不设置TTL，而是根据传送的数据来动态设定。

我们在配置类中加上QC和交换机x交换机y之间的绑定关系

@Configuration
public class DelayedQueueConfig {@Bean("queueC")public Queue queueC() {Map<String, Object> args = new HashMap<>();//设置绑定死信交换机的属性args.put("x-dead-letter-exchange", "Y");args.put("x-dead-letter-routing-key", "YD");return QueueBuilder.durable("QC").withArguments(args).build();}//绑定队列QC和交换机X之间的关系@Beanpublic Binding queueCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange) {return BindingBuilder.bind(queueC).to(xExchange).with("XC");}
}

 定义生产者

// @RestController注解表明该类是一个RESTful风格的控制器，用于处理HTTP请求并返回JSON等格式的数据
@RestController 
// @RequestMapping("ttl")注解用于映射请求路径，所有以"/ttl"开头的请求会被该控制器处理
@RequestMapping("ttl") 
public class SendMessageController { // @Resource注解用于自动装配RabbitTemplate实例，RabbitTemplate是Spring AMQP提供的用于操作RabbitMQ的工具类@Resource RabbitTemplate rabbitTemplate; // 该方法用于处理"/sendttlMessage/{message}/{ttl}"路径的请求，是消息发送的逻辑所在// @RequestMapping("sendttlMessage/{message}/{ttl}")注解将该方法映射到指定的请求路径，其中{message}和{ttl}是路径变量@RequestMapping("sendttlMessage/{message}/{ttl}") public void sendTtlMessage(@PathVariable("message") String message, @PathVariable("ttl") String ttl) { // 使用rabbitTemplate的convertAndSend方法向RabbitMQ发送消息// 第一个参数"X"指定交换机名称// 第二个参数"XC"指定路由键，用于将消息路由到绑定了该路由键的队列// 第三个参数message是消息内容// 第四个参数是一个Lambda表达式，用于在发送消息前设置消息的过期时间（通过setExpiration方法）// 设置的过期时间由路径变量ttl传入，单位为毫秒rabbitTemplate.convertAndSend("X", "XC", message, msg -> { msg.getMessageProperties().setExpiration(ttl); return msg; }); }
}