免费软件网站下载,教务系统管理系统,自己给自己网站做推广,厦门市房地产建设局网站什么是延迟队列
作用#xff1a;用来存储延迟消息延迟消息#xff1a;生产者发送一个消息给mq#xff0c;然后mq会经过一段时间#xff08;延迟时间#xff09;#xff0c;然后在把这个消息发送给消费者
应用场景
预定会议后#xff0c;需要在预定的时间点前十分钟通…
什么是延迟队列
作用用来存储延迟消息延迟消息生产者发送一个消息给mq然后mq会经过一段时间延迟时间然后在把这个消息发送给消费者
应用场景
预定会议后需要在预定的时间点前十分钟通知各个与会人员参加会议推送某些数据的定时任务微信公众号文章的延迟发布订单超时未支付自动取消订单
实现延迟队列
在rabbitmq中没有提供真正意义上的延迟队列。要实现延迟队列有两套方案
方案一基于死信队列中的消息TTL过期模式的进行改造不监听对应队列使消息过期后全部进入死信队列以达成延时效果主要有队列TTL和消息TTL两种方案二使用延时队列插件让交换机管理消息延时时间常用 创建工程
创建springBoot工程勾选需要的依赖添加RabbitMQ配置
spring.rabbitmq.hostxxxx
spring.rabbitmq.usernameguest
spring.rabbitmq.passwordguest
spring.rabbitmq.virtual-hostDeadQueue使用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(){MapString, 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(){MapString, 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);}//声明死信队列 QDBean(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恰好又设置了积压上限无法被积压的消息将直接进入延时队列达不到延时效果修改配置文件 //声明队列 QCBeanpublic Queue queueC(){MapString, 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);}生产者代码 //声明队列 QCBeanpublic Queue queueC(){MapString, 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(){MapString,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
