一种多策略下RabbitMQ的延时队列实现

1.为什么会用到延时队列?

场景: 最近在开发一款系统中遇到这样一个场景,A系统开通套餐需要把套餐信息以邮件的形式发送给相关工作人员,经过人工审核通过后,在B系统里面开通,A系统会调B系统套餐列表接口查询套餐是否开通成功,开通成功则从A系统去完成订单,假如超过设定时间未开通成功,则关闭订单并退费.
(这东西俗称"套娃")
这时候用RabbitMQ的延时队列就可以完美的解决这个问题

2.为什么会提到多策略?

场景: 假如A系统还有别的功能添加需要经过人工审核之后在B系统中添加成功之后,A系统才会显示添加成功,但是又不想写很多队列啊消费者等代码.就可以用到这种策略模式,换句话说 就是类似 if… else …能明白了吧.

3.进入今天主题

整体流程图:
在这里插入图片描述

生产者生产一条延时消息，根据需要延时时间的不同，利用routingkey将消息路由到延时队列，队列都设置了TTL属性，并绑定到死信交换机中，消息过期后，根据routingkey又会被路由到死信队列中，消费者只需要监听死信队列,拿到消息去具体的策略实现类进行后续业务处理即可。

有了这个图写代码就简单了.
mq配置类 声明队列,路由键,交换机之间的关系;以及生产者消费者 rabbitmq等Bean

RabbitMqConfig

注意 监听我也写在配置类里面SimpleMessageListenerContainer用的这个类去设置的队列
simpleMessageListenerContainer.setQueueNames(DEAD_LETTER_QUEUE_NAME);

package com.king.alice.rabbitmq.config;import com.king.alice.rabbitmq.delay.consumer.MessageConsumer;
import com.king.alice.rabbitmq.delay.consumer.Strategy;
import com.king.alice.rabbitmq.delay.provider.MessageProvider;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.List;
import java.util.Map;/*** @Author wlt* @Description rabbitmq配置类* @Date 2022/9/4* @Param* @return**/@Configuration
public class RabbitMqConfig {public static final String DELAY_EXCHANGE_NAME = "delay.alice.exchange";public static final String DELAY_QUEUE_NAME = "delay.alice.queue";public static final String DELAY_QUEUE_ROUTING_KEY = "delay.alice.queue.routing.key";public static final String DEAD_LETTER_EXCHANGE = "ttl.alice.exchange";public static final String DEAD_LETTER_QUEUE_ROUTING_KEY = "ttl.alice.queue.routing.key";public static final String DEAD_LETTER_QUEUE_NAME = "ttl.alice.queue";/*** 声明延时Exchange* @return*/@Bean("delayExchange")public DirectExchange delayExchange(){return new DirectExchange(DELAY_EXCHANGE_NAME);}/*** 功能描述: <br>* <声明死信Exchange>*/@Bean("deadLetterExchange")public DirectExchange deadLetterExchange(){return new DirectExchange(DEAD_LETTER_EXCHANGE);}/*** 声明延时队列 并绑定到对应的死信交换机* @return*/@Bean("delayQueue")public Queue delayQueue(){Map<String, Object> args = new HashMap<>(2);// x-dead-letter-exchange    这里声明当前队列绑定的死信交换机args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);// x-dead-letter-routing-key  这里声明当前队列的死信路由keyargs.put("x-dead-letter-routing-key", DEAD_LETTER_QUEUE_ROUTING_KEY);// x-message-ttl  声明队列的TTL
//        args.put("x-message-ttl", 6000);return QueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();}/*** 功能描述: <br>* <声明死信队列用于接收延时处理的消息>*/@Bean("deadLetterQueue")public Queue deadLetterQueue(){return new Queue(DEAD_LETTER_QUEUE_NAME);}/*** 功能描述: <br>* <声明延时队列绑定关系>* @Param:* @Return:* @Author: 大魔王* @Date: 2023/8/15 20:00*/@Beanpublic Binding delayBinding(@Qualifier("delayQueue") Queue queue,@Qualifier("delayExchange") DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY);}/*** 功能描述: <br>* <声明死信队列A绑定关系>* @Param:* @Return:* @Author: 大魔王* @Date: 2023/8/15 20:01*/@Beanpublic Binding deadLetterBinding(@Qualifier("deadLetterQueue") Queue queue,@Qualifier("deadLetterExchange") DirectExchange exchange){return BindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUE_ROUTING_KEY);}@Bean@ConditionalOnMissingBeanpublic MessageProvider messageProvider(@Qualifier("delayRabbitTemplate") RabbitTemplate template) {return new MessageProvider(template);}@Bean@ConditionalOnMissingBeanpublic MessageConsumer messageConsumer(ObjectProvider<List<Strategy>> provider) {return new MessageConsumer(provider);}@Bean@ConditionalOnMissingBeanpublic RabbitTemplate delayRabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template = new RabbitTemplate(connectionFactory);template.setMessageConverter(new Jackson2JsonMessageConverter());return template;}@BeanSimpleMessageListenerContainer simpleMessageListenerContainer(MessageConsumer messageConsumer, ConnectionFactory factory) {SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(factory);simpleMessageListenerContainer.setQueueNames(DEAD_LETTER_QUEUE_NAME);simpleMessageListenerContainer.setExposeListenerChannel(true);simpleMessageListenerContainer.setMessageListener(messageConsumer);return simpleMessageListenerContainer;}public static final String EXCHANGE_NAME = "alice_topic_exchange";public static final String QUEUE_NAME = "alice_queue";@Bean("aliceExchange")public Exchange aliceExchange() {return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}@Bean("aliceQueue")public Queue aliceQueue() {return QueueBuilder.durable(QUEUE_NAME).build();}@Beanpublic Binding bindQueueExchange(@Qualifier("aliceQueue") Queue queue, @Qualifier("aliceExchange") Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("alice.#").noargs();}}

生产者:

MessageProvider

package com.king.alice.rabbitmq.delay.provider;import cn.hutool.core.date.DateUtil;
import com.king.alice.common.json.JSON;
import com.king.alice.rabbitmq.config.RabbitMqConfig;
import com.king.alice.rabbitmq.delay.bean.DelayMessage;
import lombok.extern.slf4j.Slf4j;
import org.junit.Assert;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.util.Date;/*** @author 大魔王*/
@Slf4j
@Component
public class MessageProvider {@Autowiredprivate final RabbitTemplate rabbitTemplate;public MessageProvider(RabbitTemplate rabbitTemplate) {this.rabbitTemplate = rabbitTemplate;}/*** send delay message*/public void sendMessage(DelayMessage delayMessage) {Assert.assertNotNull(delayMessage);log.info(" now date {},delay {} seconds to write to the message queue", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"), delayMessage.getDelay());rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE_NAME, RabbitMqConfig.DELAY_QUEUE_ROUTING_KEY, delayMessage,message -> {message.getMessageProperties().setExpiration(String.valueOf(delayMessage.getDelay() * 1000));return message;});}}

消费者:

package com.king.alice.rabbitmq.delay.consumer;import cn.hutool.core.util.ObjectUtil;
import com.king.alice.common.json.JSONObject;
import com.king.alice.rabbitmq.delay.bean.AliceMessage;
import com.king.alice.rabbitmq.delay.bean.DelayMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;/*** @author 大魔王*/
@Slf4j
public class MessageConsumer implements MessageListener {private final Map<Type, List<Strategy>> strategyMap = new ConcurrentHashMap<>();public MessageConsumer(ObjectProvider<List<Strategy>> stategyProvider) {List<Strategy> handleList = stategyProvider.getIfAvailable();Optional<? extends List<Strategy>> optionalStrategies = Optional.ofNullable(handleList);optionalStrategies.ifPresent(strategies -> strategies.stream().filter(strategy -> {Type genericInterface = strategy.getClass().getGenericInterfaces()[0];return genericInterface instanceof ParameterizedType;}).map(strategy -> ((ParameterizedType) strategy.getClass().getGenericInterfaces()[0]).getActualTypeArguments()[0]).collect(Collectors.toSet()).forEach(delayMessages -> {List<Strategy> collect = strategies.stream().filter(strategy -> {Type genericInterface = strategy.getClass().getGenericInterfaces()[0];if (genericInterface instanceof ParameterizedType) {Type actualTypeArgument = ((ParameterizedType) genericInterface).getActualTypeArguments()[0];return delayMessages.getTypeName().equals(actualTypeArgument.getTypeName());}return false;}).collect(Collectors.toList());strategyMap.put(delayMessages, collect);}));}@Overridepublic void onMessage(Message message) {MessageConverter messageConverter = new Jackson2JsonMessageConverter();DelayMessage delayMessage = (DelayMessage) messageConverter.fromMessage(message);List<Strategy> strategyList = strategyMap.get(delayMessage.getClass());if (!CollectionUtils.isEmpty(strategyList)) {strategyList.forEach(strategy -> strategy.handle(delayMessage));} else {log.info("Missing message processing class");}}}

策略相关Bean,接口以及实现类

DelayMessage

package com.king.alice.rabbitmq.delay.bean;/*** @author 大魔王*/
public interface DelayMessage{/*** 获得延迟时间（单位秒）** @return 延迟时间单位秒*/int getDelay();}

AliceMessage

package com.king.alice.rabbitmq.delay.bean;import lombok.Getter;
import lombok.Setter;/*** @author 大魔王*/
@Getter
@Setter
public class AliceMessage implements DelayMessage {/***  用户邮箱*/String email;/***  订单类型*/String orderType;/***  执行次数*/Integer dealCount;/***  延时秒数*/int delay;@Overridepublic int getDelay() {return this.delay;}public void setDelay(int delay) {this.delay = delay;}
}

UserMessage

package com.king.alice.rabbitmq.delay.bean;import lombok.Getter;
import lombok.Setter;/*** @author 大魔王*/
@Getter
@Setter
public class UserMessage implements DelayMessage{/***  用户*/String username;/***  token*/String token;/***  执行次数*/Integer dealCount;/***  延时秒数*/int delay;@Overridepublic int getDelay() {return this.delay;}public void setDelay(int delay) {this.delay = delay;}
}

Strategy

package com.king.alice.rabbitmq.delay.consumer;import com.king.alice.rabbitmq.delay.bean.DelayMessage;/*** @author 大魔王*/
public interface Strategy<T extends DelayMessage> {/*** 处理消息的方法** @param delayMessage 收到的消息*/void handle(T delayMessage);
}

AliceMessageHandler

package com.king.alice.rabbitmq;import com.king.alice.common.json.JSON;
import com.king.alice.common.json.JSONObject;
import com.king.alice.rabbitmq.delay.bean.AliceMessage;
import com.king.alice.rabbitmq.delay.consumer.Strategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;/*** @author 大魔王*/
@Component
@Slf4j
public class AliceMessageHandler implements Strategy<AliceMessage> {@Overridepublic void handle(AliceMessage delayMessage) {log.info("AliceMessage响应体{}", JSONObject.parseObject(JSON.toJSONString(delayMessage)));}
}

UserMessageHandler

package com.king.alice.rabbitmq;import com.king.alice.common.json.JSON;
import com.king.alice.common.json.JSONObject;
import com.king.alice.rabbitmq.delay.bean.UserMessage;
import com.king.alice.rabbitmq.delay.consumer.Strategy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;/*** @author 大魔王*/
@Slf4j
@Component
public class UserMessageHandler implements Strategy<UserMessage> {@Overridepublic void handle(UserMessage delayMessage) {log.info("UserMessage响应体{}", JSONObject.parseObject(JSON.toJSONString(delayMessage)));}
}

接下来 我们写个controller测试一下

SysAccountController

package com.king.alice.manage.sys.controller;import cn.hutool.core.date.DateUtil;
import com.king.alice.manage.sys.entity.SysAccount;
import com.king.alice.manage.sys.service.SysAccountService;
import com.king.alice.rabbitmq.delay.bean.AliceMessage;
import com.king.alice.rabbitmq.delay.bean.UserMessage;
import com.king.alice.rabbitmq.delay.provider.MessageProvider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;import javax.annotation.Resource;
import java.util.Date;/*** 账号表(SysAccount)表控制层** @author makejava* @since 2023-08-09 11:40:16*/
@RestController
@Slf4j
public class SysAccountController {/*** 服务对象*/@Resourceprivate SysAccountService sysAccountService;@Autowiredprivate MessageProvider messageProvider;@PostMapping("/send-alice-message")public String sendMsg(@RequestBody AliceMessage aliceMessage) {messageProvider.sendMessage(aliceMessage);log.info("当前时间：{},收到aliceMessage请求，msg:{}", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"), aliceMessage);return "success";}@PostMapping("/send-user-message")public String sendMsg(@RequestBody UserMessage userMessage) {messageProvider.sendMessage(userMessage);log.info("当前时间：{},收到userMessage请求，msg:{}", DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"), userMessage);return "success";}
}

调接口

第一个策略:
在这里插入图片描述
控制台打印
在这里插入图片描述
第二个策略:
在这里插入图片描述
延时12秒成功接收到消息
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/69835.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

zabbix监控Nginx

zabbix监控Nginx

目录 一、环境准备 二、部署Nginx被监控端 三、自定义Nginx监控key 四、给目标主机创建监控项 一、环境准备 搭建zabbix基础环境 zabbix基础环境部署参照&#xff1a;部署Zabbix监控 以下实验部署均基于上述环境 二、部署Nginx被监控端 此处选择在web1搭建Nginx服务&am…
阅读更多...
HCIP学习--BGP实验

HCIP学习--BGP实验

一、实验拓扑 二、实验需求 除R5的5.5.5.0环回外&#xff0c;其他所有的环回均可互相访问 三、实验步骤 首先配置IP&#xff0c;配置好IBGP 建立直连的EBGP邻居关系 R1和R2建立直连的EBGP邻居关系 [r1]bgp 1 [r1-bgp]router-id 1.1.1.1 [r1-bgp]peer 12.1.1.2 as-number …
阅读更多...
剑指offer-2.2字符串

剑指offer-2.2字符串

字符串 C/C中每个字符串都以字符"\0"作为结尾&#xff0c;这样我们就能很方便地找到字符串的最后尾部。但由于这个特点&#xff0c;每个字符串中都有一个额外字符的开销&#xff0c;稍不留神就会造成字符串的越界。比如下面的代码&#xff1a; char str [10]; strc…
阅读更多...
编程练习（1）

编程练习（1）

目录 一.选择题 第一题&#xff1a; 第二题&#xff1a; 第三题&#xff1a; 第四题&#xff1a; 第五题&#xff1a; ​编辑 二.编程题 第一题&#xff1a; 第二题&#xff1a; 1.暴力方法&#xff1a; 2.数组法&#xff1a; 一.选择题 第一题&#xff1a; 解析&…
阅读更多...
已知四个坐标点，怎样求出四边形的四个内角

已知四个坐标点，怎样求出四边形的四个内角

1&#xff0c;理论 最简单的方式利用向量进行求解 如图可得&#xff1a; cosθa*b/&#xff08;|a|*|b|&#xff09; 已知三点坐标&#xff0c;很容易可以得到两向量之积a*b&#xff0c;以及每个的模值 2&#xff0c;四个角度求解过程 首先&#xff0c;我们定义了四个坐标点…
阅读更多...
微服务学习笔记-基本概念

微服务学习笔记-基本概念

微服务是一种经过良好架构设计的分布式架构方案。根据业务功能对系统做拆分&#xff0c;每个业务功能模块作为独立项目开发&#xff0c;称为一个服务。 微服务的架构特征&#xff1a; 单一职责&#xff1a;微服务拆分粒度更小&#xff0c;每一个服务都对应唯一的业务能力&…
阅读更多...
Vivado使用入门之二:网表物理约束

Vivado使用入门之二:网表物理约束

目录 一、背景 二、物理约束 2.1 概念 2.2 网表约束 2.2.1 CLOCK_DEDICATED_ROUTE 2.2.2 MARK_DEBUG 2.2.3 DONT_TOUCH 2.2.4 LOCK_PINS 三、位置约束 四、布线约束 4.1 route 4.2 assign routing mode 五、参考 一、背景 在工程设计中为了保证上板后功能正常&…
阅读更多...
RabbitMQ工作流程详解

RabbitMQ工作流程详解

1 生产者发送消息的流程 (1)生产者连接RabbitMQ&#xff0c;建立TCP连接(Connection)&#xff0c;开启信道(Channel) (2)生产者声明一个Exchange (交换器)&#xff0c;并设置相关属性&#xff0c;比如交换器类型、是否持久化等 (3)生产者声明一个队列井设置相关属性&#xf…
阅读更多...
eachars 自适应

eachars 自适应

目录 1. 案例&#xff1a; 2. 原因&#xff1a; 3. 解决&#xff1a; 1. 案例&#xff1a; 默认是正常宽度&#xff08;如图1&#xff09;&#xff0c;当再次跳转会该页面时&#xff0c;eachars图发生变化&#xff08;如图2&#xff09;。 图1 图2 2. 原因&#xff1a; 没有…
阅读更多...
Gitlab-第四天-CD到k8s集群的坑

Gitlab-第四天-CD到k8s集群的坑

一、.gitlab-ci.yml #CD到k8s集群的 stages: - deploy-test build-image-deploy-test: stage: deploy-test image: bitnami/kubectl:latest # 使用一个包含 kubectl 工具的镜像 tags: - k8s script: - ls -al - kubectl apply -f deployment.yaml # 根据实际情况替换…
阅读更多...
Redis持久化——RDB和AOF

Redis持久化——RDB和AOF

Redis数据库是内存数据库&#xff0c;一旦出现服务宕机&#xff0c;那么内存中的数据就容易丢失。所以需要进行redis的持久化动作。 Redis持久化是指将Redis内存数据持存储到磁盘中&#xff0c;若出现了Redis服务宕机后&#xff0c;能够从硬盘中再恢复到Redis内存中。 Redis的持…
阅读更多...
公司电脑三维图纸加密、机械图挡加密软件

公司电脑三维图纸加密、机械图挡加密软件

机械图纸加密软件的问世&#xff0c;让很多的网络公司都大受其带来的工作中的便利。在安装了机械图纸加密软件后&#xff0c;不仅可以很好的管理员工在工作时的上网娱乐&#xff0c;在对整个公司员工的工作效率上也有着明显的提高&#xff0c;那么对于机械图纸加密软件的具体特…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023