网页设计师网站国家职业技能培训平台

什么是延迟队列

作用：用来存储延迟消息
延迟消息：生产者发送一个消息给mq，然后mq会经过一段时间（延迟时间），然后在把这个消息发送给消费者

应用场景

1. 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议
2. 推送某些数据的定时任务
3. 微信公众号文章的延迟发布
4. 订单超时未支付自动取消订单
   

实现延迟队列

在rabbitmq中没有提供真正意义上的延迟队列。要实现延迟队列有两套方案

1. 方案一：基于死信队列中的消息TTL过期模式的进行改造，不监听对应队列，使消息过期后全部进入死信队列以达成延时效果，主要有队列TTL和消息TTL两种
2. 方案二：使用延时队列插件，让交换机管理消息延时时间（常用）

---

创建工程

创建springBoot工程，勾选需要的依赖

添加RabbitMQ配置

spring.rabbitmq.host=xxxx
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=DeadQueue

使用TTL+死信队列

队列TTL案例

对队列QA设置过期时间 10S，队列QB设置过期时间 40S，不监听QA、QB队列，使消息进入队列后不被消费导致TTL超时进入QD延迟队列

Y是死信交换机，QD是死信队列

缺点：每增加一个新的时间需求，就要新增一个队列
创建RabbitMQ配置文件

package com.dmbjz.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/* RabbitMQ的交换机、队列配置文件 */
@Configuration
public class ExchangeQueueConfig {public static final String X_EXCHANGE = "X";public static final String QUEUE_A = "QA";public static final String QUEUE_B = "QB";public static final String Y_DEAD_LETTER_EXCHANGE = "Y";public static final String DEAD_LETTER_QUEUE = "QD";/*创建X交换机*/@Beanpublic DirectExchange xExchange(){return new DirectExchange(X_EXCHANGE);}/*创建死信交换机*/@Beanpublic DirectExchange yExchange(){return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);}//声明队列 A ttl 为 10s 并绑定到对应的死信交换机@Bean("queueA")public Queue queueA(){Map<String, Object> args = new HashMap<>(3);args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 keyargs.put("x-message-ttl", 10000);                                //声明队列的 TTLreturn QueueBuilder.durable(QUEUE_A).withArguments(args).build();}// 声明队列 A 绑定 X 交换机@Beanpublic Binding queueaBindingX(@Qualifier("queueA") Queue queueA,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueA).to(xExchange).with("XA");}//声明队列 B ttl 为 40s 并绑定到对应的死信交换机@Bean("queueB")public Queue queueB(){Map<String, Object> args = new HashMap<>(3);args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);         //声明当前队列绑定的死信交换机args.put("x-dead-letter-routing-key", "YD");                        //声明当前队列的死信路由 keyargs.put("x-message-ttl", 40000);                                   //声明队列的 TTLreturn QueueBuilder.durable(QUEUE_B).withArguments(args).build();}//声明队列 B 绑定 X 交换机@Beanpublic Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queue1B).to(xExchange).with("XB");}//声明死信队列 QD@Bean("queueD")public Queue queueD(){return new Queue(DEAD_LETTER_QUEUE);}//声明死信队列 QD 绑定关系@Beanpublic Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,@Qualifier("yExchange") DirectExchange yExchange){return BindingBuilder.bind(queueD).to(yExchange).with("YD");}
}

生产者代码：

package com.dmbjz.controller;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;
import java.util.Date;/* 生产者发送消息Controller */
@RestController
@RequestMapping("/ttl")
@Slf4j
public class SendMessageController {@Autowiredprivate RabbitTemplate rabbitTemplate;@RequestMapping("/sendMessage/{message}")public void sendMsg(@PathVariable String message){log.info("当前时间:{},发送一条信息给两个TTL队列,消息内容:{}",new Date(),message);rabbitTemplate.convertAndSend("X","XA",message.getBytes(StandardCharsets.UTF_8));rabbitTemplate.convertAndSend("X","XB",message.getBytes(StandardCharsets.UTF_8));}
}

消费者代码：

package com.dmbjz.consumer;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Date;/* 队列TTL消费者 */
@Component
@Slf4j
public class DeadLetterQueueConsumer {@RabbitListener(queues = "QD")public void receiveD(Message message, Channel channel)throws Exception{String msg = new String(message.getBody());log.info("当前时间:{},收到死信队列的消息:{}",new Date(),msg);}
}

浏览器访问地址测试：

http://localhost:8080/ttl/sendMessage/测试消息TTL

---

消息TTL案例

对消息设置过期时间，不监听QC队列，消息超时后自动进入QD延迟队列
缺点：如果积压在队列前面的消息延时时长很长，而后面积压的消息延时时长很短，积压时间短的消息并不会被提前放入死信队列；如果QC恰好又设置了积压上限，无法被积压的消息将直接进入延时队列，达不到延时效果

修改配置文件：

    //声明队列 QC@Beanpublic Queue queueC(){Map<String, Object> args = new HashMap<>(3);args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 keyreturn QueueBuilder.durable(QUEUE_C).withArguments(args).build();}//声明队列 QC 绑定 X 交换机@Beanpublic Binding queuebCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueC()).to(xExchange).with("XC");}

生产者代码：

    //声明队列 QC@Beanpublic Queue queueC(){Map<String, Object> args = new HashMap<>(3);args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);      //声明当前队列绑定的死信交换机args.put("x-dead-letter-routing-key", "YD");                     //声明当前队列的死信路由 keyreturn QueueBuilder.durable(QUEUE_C).withArguments(args).build();}//声明队列 QC 绑定 X 交换机@Beanpublic Binding queuebCBindingX(@Qualifier("queueC") Queue queueC,@Qualifier("xExchange") DirectExchange xExchange){return BindingBuilder.bind(queueC()).to(xExchange).with("XC");}

浏览器访问地址进行测试：

http://localhost:8080/ttl/sendMessagExpira/测试消息1/10000
http://localhost:8080/ttl/sendMessagExpira/测试消息2/1000

---

延时插件

使用延时队列插件实现延时队列功能，原理为交换机管理消息延时时间
插件版本需要兼容 RabbitMQ 版本，具体参考其发布说明**，**延时队列插件下载：github
插件安装步骤

1.将安装目录的延时队列插件拷贝到RabbitMQ插件目录cp rabbitmq_delayed_message_exchange-3.8.0.ez /root/rabbitmq_server-3.8.8/plugins2.安装延时队列插件   rabbitmq-plugins enable rabbitmq_delayed_message_exchange3、重启RabbitMQ服务systemctl restart rabbitmq-server

案例演示：
延时队列插件实际落地固定为图中架构模式

创建配置文件：

package com.dmbjz.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/* 延时队列插件案例 RabbitMQ配置类 */
@Configuration
public class DelayedQueueConfig {private static final String delayed_queue_name = "delayed.queue";private static final String delayed_exchange_name = "delayed.exchange";private static final String delayed_routingkey = "delayed.routingkey";/*创建延时插件的交换机，需要使用自定义方法进行创建*   插件版非死信队列，不需要路由到不同的交换机进行指定过期时间，所以固定为 direct 类型交换机* */@Beanpublic CustomExchange delayedExchange(){Map<String,Object> map = new HashMap<>(1);map.put("x-delayed-type","direct");       //延迟队列类型，固定值return new CustomExchange(delayed_exchange_name,"x-delayed-message",true,false,map);}/*队列*/@Beanpublic Queue delayQueue(){return QueueBuilder.durable(delayed_queue_name).build();}/*绑定，自定义交换机绑定多一个 noargs方法 */@Beanpublic Binding delayBing(@Qualifier("delayQueue") Queue delayQueue,@Qualifier("delayedExchange") CustomExchange delayedExchange){return BindingBuilder.bind(delayQueue).to(delayedExchange).with(delayed_routingkey).noargs();}
}

生产者代码：

    /*延时插件案例*/@RequestMapping("/sendMessagPlugin/{message}/{time}")public void sendMsgPlugin(@PathVariable String message,@PathVariable Integer time){MessageProperties properties = new MessageProperties();properties.setDelay(time);      //设置延时时间Message msg = new Message(message.getBytes(StandardCharsets.UTF_8),properties);log.info("当前时间:{},发送具有过期时间为{}毫秒的信息给延时插件队列,消息内容:{}",new Date(),time,message);rabbitTemplate.convertAndSend("delayed.exchange","delayed.routingkey",msg);}

消费者代码：

package com.dmbjz.consumer;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.util.Date;/* 延时队列插件 消费者 */
@Component
@Slf4j
public class DelayQueueConsumer {@RabbitListener(queues = "delayed.queue")public void receiveDelayQueue(Message message, Channel channel)throws Exception{String msg = new String(message.getBody());log.info("当前时间:{},收到死信队列的消息:{}",new Date(),msg);}
}

浏览器访问地址进行测试：

http://localhost:8080/ttl/sendMessagPlugin/测试消息1/10000
http://localhost:8080/ttl/sendMessagPlugin/测试消息2/1000