安徽做网站哪家好,免费简约ppt模板,网站建设鸿儒,现在做网络推广好做吗介绍
延迟队列#xff0c;队列内部是有序的#xff0c;延迟队列中的元素是希望在指定时间到了以后或之前取出和处理。
死信队列中#xff0c;消息TTL过期的情况其实就是延迟队列。
使用场景
1.订单在十分钟内未支付则自动取消。
2.新创建的店铺#xff0c;如果十天内没…介绍
延迟队列队列内部是有序的延迟队列中的元素是希望在指定时间到了以后或之前取出和处理。
死信队列中消息TTL过期的情况其实就是延迟队列。
使用场景
1.订单在十分钟内未支付则自动取消。
2.新创建的店铺如果十天内没有上传过商品则自动发送消息提醒。
3.用户注册成功后如果三天没有登陆则进行短信提醒。
4.用户发起退款如果三天内没有得到处理则通知相关运营人员。
5.预定会议后需要再预定的时间点前十分钟通知各个与会人员参加会议。
传统实现方案
可以开启一个定时任务每秒中去轮训检查取出需要被处理的数据。对于数据量较少可以这么做。但是如果有大数据量的任务需要处理活动期间达到百万级或者千万级的庞大数据量是不可取的。
整合SpringBoot
引入依赖 !--rabbitmq整合springboot使用的--
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency
!--rabbitmq测试依赖--
dependencygroupIdorg.springframework.amqp/groupIdartifactIdspring-rabbit-test/artifactIdscopetest/scope
/dependency 配置文件添加配置 spring:rabbitmq:host: 192.168.171.128username: adminpassword: 123port: 5672 延迟队列代码架构图 说明QD为死信队列QA和QB是普通队列。x为直接交换机y为死信交换机。
代码结构的转变
整合springboot项目之前将队列和交换机的声明和配置放在消费者端的。那么整合springboot之后将会将这些配置放在配置文件类代码中。
代码
配置类声明队列、交换机、绑定关系
package com.xkj.org.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;/*** TTL队列配置文件代码*/
Configuration
public class TtlQueueConfig {//普通交换机public static final String X_EXCHANGE X;//死信交换机public static final String Y_HEAD_LETTER_EXCHANGE Y;//普通队列public static final String QUEUE_A QA;public static final String QUEUE_B QB;//死信队列public static final String DEAD_LETTER_QUEUE QD;/*** 声明普通交换机X,bean的别名xExchange* return*/Bean(xExchange)public DirectExchange xExchange() {return new DirectExchange(X_EXCHANGE);}/*** 声明死信交换机Y,bean的别名yExchange* return*/Bean(yExchange)public DirectExchange yExchange() {return new DirectExchange(Y_HEAD_LETTER_EXCHANGE);}/*** 声明普通队列QA* return*/Bean(queueA)public Queue queueA() {MapString, Object arguments new HashMap(3);//设置死信交换机arguments.put(x-dead-letter-exchange, Y_HEAD_LETTER_EXCHANGE);//声明死信的routingeyarguments.put(x-dead-letter-routing-key, YD);//设置消息过期时间ttl为10sarguments.put(x-message-ttl, 10000);return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();}/*** 声明普通队列QB* return*/Bean(queueB)public Queue queueB() {MapString, Object arguments new HashMap(3);//设置死信交换机arguments.put(x-dead-letter-exchange, Y_HEAD_LETTER_EXCHANGE);//声明死信的routingKeyarguments.put(x-dead-letter-routing-key, YD);//设置消息过期时间ttl为40sarguments.put(x-message-ttl, 40000);return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();}/*** 声明死信队列QD* return*/Bean(queueD)public Queue queueD() {return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();}/*** 将队列QA绑定到交换机X上指定routingKey为XA* param queueA* param xExchange* return*/Beanpublic Binding queueABindingX(Qualifier(queueA) Queue queueA, Qualifier(xExchange)DirectExchange xExchange) {return BindingBuilder.bind(queueA).to(xExchange).with(XA);}/*** 将队列QB绑定到交换机X上指定routingKey为XB* param queueB* param xExchange* return*/Beanpublic Binding queueBBindingX(Qualifier(queueB) Queue queueB, Qualifier(xExchange) DirectExchange xExchange) {return BindingBuilder.bind(queueB).to(xExchange).with(XB);}/*** 将队列QD绑定到交换机Y上指定routingKey为YD* param queueD* param yExchange* return*/Beanpublic Binding queueDBindingY(Qualifier(queueD) Queue queueD, Qualifier(yExchange) DirectExchange yExchange) {return BindingBuilder.bind(queueD).to(yExchange).with(YD);}
}生产者通过controller接口发送
package com.xkj.org.controller;import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
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.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;/*** 消息生产者*/
Slf4j
RestController
RequestMapping(/ttl)
Api(tags 消息生产者, description 消息生产者控制器)
public class MessageController {Autowiredprivate RabbitTemplate rabbitTemplate;ApiOperation(消息发送测试)GetMapping(/sendMsg/{msg})public void sendMsg(ApiParam(value 发送的消息内容, required true) PathVariable(msg) String message) {log.info(当前时间{}发送一条消息给两个队列{}, new Date().toString(), message);rabbitTemplate.convertAndSend(X, XA, ttl10s的消息 message);rabbitTemplate.convertAndSend(X, XB, ttl40s的消息 message);}}消费者监听器
package com.xkj.org.listener;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;/*** 消费者*/
Slf4j
Componentpublic class DeadLetterQueueConsumer {RabbitListener(queues QD)public void receiveD(Message message, Channel channel) throws Exception {String msg new String(message.getBody(), UTF-8);log.info(当前时间{}收到死信队列的消息{}, new Date().toString(), msg);}
}结果
