0
0

一种多策略下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);

packagecom.king.alice.rabbitmq.config;importcom.king.alice.rabbitmq.delay.consumer.MessageConsumer;importcom.king.alice.rabbitmq.delay.consumer.Strategy;importcom.king.alice.rabbitmq.delay.provider.MessageProvider;importorg.springframework.amqp.core.*;importorg.springframework.amqp.rabbit.connection.ConnectionFactory;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;importorg.springframework.amqp.support.converter.Jackson2JsonMessageConverter;importorg.springframework.beans.factory.ObjectProvider;importorg.springframework.beans.factory.annotation.Qualifier;importorg.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;importjava.util.HashMap;importjava.util.List;importjava.util.Map;/**
 * @Author wlt
 * @Description rabbitmq配置类
 * @Date 2022/9/4
 * @Param
 * @return
 **/@ConfigurationpublicclassRabbitMqConfig{publicstaticfinalStringDELAY_EXCHANGE_NAME="delay.alice.exchange";publicstaticfinalStringDELAY_QUEUE_NAME="delay.alice.queue";publicstaticfinalStringDELAY_QUEUE_ROUTING_KEY="delay.alice.queue.routing.key";publicstaticfinalStringDEAD_LETTER_EXCHANGE="ttl.alice.exchange";publicstaticfinalStringDEAD_LETTER_QUEUE_ROUTING_KEY="ttl.alice.queue.routing.key";publicstaticfinalStringDEAD_LETTER_QUEUE_NAME="ttl.alice.queue";/**
     * 声明延时Exchange
     * @return
     */@Bean("delayExchange")publicDirectExchangedelayExchange(){returnnewDirectExchange(DELAY_EXCHANGE_NAME);}/**
     * 功能描述: <br>
     * <声明死信Exchange>
     */@Bean("deadLetterExchange")publicDirectExchangedeadLetterExchange(){returnnewDirectExchange(DEAD_LETTER_EXCHANGE);}/**
     * 声明延时队列 并绑定到对应的死信交换机
     * @return
     */@Bean("delayQueue")publicQueuedelayQueue(){Map<String,Object> args =newHashMap<>(2);// x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange",DEAD_LETTER_EXCHANGE);// x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key",DEAD_LETTER_QUEUE_ROUTING_KEY);// x-message-ttl  声明队列的TTL//        args.put("x-message-ttl", 6000);returnQueueBuilder.durable(DELAY_QUEUE_NAME).withArguments(args).build();}/**
     * 功能描述: <br>
     * <声明死信队列用于接收延时处理的消息>
     */@Bean("deadLetterQueue")publicQueuedeadLetterQueue(){returnnewQueue(DEAD_LETTER_QUEUE_NAME);}/**
     * 功能描述: <br>
     * <声明延时队列绑定关系>
     * @Param:
     * @Return:
     * @Author: 大魔王
     * @Date: 2023/8/15 20:00
     */@BeanpublicBindingdelayBinding(@Qualifier("delayQueue")Queue queue,@Qualifier("delayExchange")DirectExchange exchange){returnBindingBuilder.bind(queue).to(exchange).with(DELAY_QUEUE_ROUTING_KEY);}/**
     * 功能描述: <br>
     * <声明死信队列A绑定关系>
     * @Param:
     * @Return:
     * @Author: 大魔王
     * @Date: 2023/8/15 20:01
     */@BeanpublicBindingdeadLetterBinding(@Qualifier("deadLetterQueue")Queue queue,@Qualifier("deadLetterExchange")DirectExchange exchange){returnBindingBuilder.bind(queue).to(exchange).with(DEAD_LETTER_QUEUE_ROUTING_KEY);}@Bean@ConditionalOnMissingBeanpublicMessageProvidermessageProvider(@Qualifier("delayRabbitTemplate")RabbitTemplate template){returnnewMessageProvider(template);}@Bean@ConditionalOnMissingBeanpublicMessageConsumermessageConsumer(ObjectProvider<List<Strategy>> provider){returnnewMessageConsumer(provider);}@Bean@ConditionalOnMissingBeanpublicRabbitTemplatedelayRabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate template =newRabbitTemplate(connectionFactory);
        template.setMessageConverter(newJackson2JsonMessageConverter());return template;}@BeanSimpleMessageListenerContainersimpleMessageListenerContainer(MessageConsumer messageConsumer,ConnectionFactory factory){SimpleMessageListenerContainer simpleMessageListenerContainer =newSimpleMessageListenerContainer(factory);
        simpleMessageListenerContainer.setQueueNames(DEAD_LETTER_QUEUE_NAME);
        simpleMessageListenerContainer.setExposeListenerChannel(true);
        simpleMessageListenerContainer.setMessageListener(messageConsumer);return simpleMessageListenerContainer;}publicstaticfinalStringEXCHANGE_NAME="alice_topic_exchange";publicstaticfinalStringQUEUE_NAME="alice_queue";@Bean("aliceExchange")publicExchangealiceExchange(){returnExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}@Bean("aliceQueue")publicQueuealiceQueue(){returnQueueBuilder.durable(QUEUE_NAME).build();}@BeanpublicBindingbindQueueExchange(@Qualifier("aliceQueue")Queue queue,@Qualifier("aliceExchange")Exchange exchange){returnBindingBuilder.bind(queue).to(exchange).with("alice.#").noargs();}}

生产者:

MessageProvider

packagecom.king.alice.rabbitmq.delay.provider;importcn.hutool.core.date.DateUtil;importcom.king.alice.common.json.JSON;importcom.king.alice.rabbitmq.config.RabbitMqConfig;importcom.king.alice.rabbitmq.delay.bean.DelayMessage;importlombok.extern.slf4j.Slf4j;importorg.junit.Assert;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;importjava.util.Date;/**
 * @author 大魔王
 */@Slf4j@ComponentpublicclassMessageProvider{@AutowiredprivatefinalRabbitTemplate rabbitTemplate;publicMessageProvider(RabbitTemplate rabbitTemplate){this.rabbitTemplate = rabbitTemplate;}/**
     * send delay message
     */publicvoidsendMessage(DelayMessage delayMessage){Assert.assertNotNull(delayMessage);
        log.info(" now date {},delay {} seconds to write to the message queue",DateUtil.format(newDate(),"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;});}}

消费者:

packagecom.king.alice.rabbitmq.delay.consumer;importcn.hutool.core.util.ObjectUtil;importcom.king.alice.common.json.JSONObject;importcom.king.alice.rabbitmq.delay.bean.AliceMessage;importcom.king.alice.rabbitmq.delay.bean.DelayMessage;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.core.AcknowledgeMode;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.core.MessageListener;importorg.springframework.amqp.support.converter.Jackson2JsonMessageConverter;importorg.springframework.amqp.support.converter.MessageConverter;importorg.springframework.beans.factory.ObjectProvider;importorg.springframework.stereotype.Component;importorg.springframework.util.CollectionUtils;importjava.lang.reflect.ParameterizedType;importjava.lang.reflect.Type;importjava.util.List;importjava.util.Map;importjava.util.Optional;importjava.util.concurrent.ConcurrentHashMap;importjava.util.stream.Collectors;/**
 * @author 大魔王
 */@Slf4jpublicclassMessageConsumerimplementsMessageListener{privatefinalMap<Type,List<Strategy>> strategyMap =newConcurrentHashMap<>();publicMessageConsumer(ObjectProvider<List<Strategy>> stategyProvider){List<Strategy> handleList = stategyProvider.getIfAvailable();Optional<?extendsList<Strategy>> optionalStrategies =Optional.ofNullable(handleList);
        optionalStrategies.ifPresent(strategies -> strategies.stream().filter(strategy ->{Type genericInterface = strategy.getClass().getGenericInterfaces()[0];return genericInterface instanceofParameterizedType;}).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 instanceofParameterizedType){Type actualTypeArgument =((ParameterizedType) genericInterface).getActualTypeArguments()[0];return delayMessages.getTypeName().equals(actualTypeArgument.getTypeName());}returnfalse;}).collect(Collectors.toList());
                    strategyMap.put(delayMessages, collect);}));}@OverridepublicvoidonMessage(Message message){MessageConverter messageConverter =newJackson2JsonMessageConverter();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

packagecom.king.alice.rabbitmq.delay.bean;/**
 * @author 大魔王
 */publicinterfaceDelayMessage{/**
     * 获得延迟时间（单位秒）
     *
     * @return 延迟时间单位秒
     */intgetDelay();}

AliceMessage

packagecom.king.alice.rabbitmq.delay.bean;importlombok.Getter;importlombok.Setter;/**
 * @author 大魔王
 */@Getter@SetterpublicclassAliceMessageimplementsDelayMessage{/**
     *  用户邮箱
     */String email;/**
     *  订单类型
     */String orderType;/**
     *  执行次数
     */Integer dealCount;/**
     *  延时秒数
     */int delay;@OverridepublicintgetDelay(){returnthis.delay;}publicvoidsetDelay(int delay){this.delay = delay;}}

UserMessage

packagecom.king.alice.rabbitmq.delay.bean;importlombok.Getter;importlombok.Setter;/**
 * @author 大魔王
 */@Getter@SetterpublicclassUserMessageimplementsDelayMessage{/**
     *  用户
     */String username;/**
     *  token
     */String token;/**
     *  执行次数
     */Integer dealCount;/**
     *  延时秒数
     */int delay;@OverridepublicintgetDelay(){returnthis.delay;}publicvoidsetDelay(int delay){this.delay = delay;}}

Strategy

packagecom.king.alice.rabbitmq.delay.consumer;importcom.king.alice.rabbitmq.delay.bean.DelayMessage;/**
 * @author 大魔王
 */publicinterfaceStrategy<TextendsDelayMessage>{/**
     * 处理消息的方法
     *
     * @param delayMessage 收到的消息
     */voidhandle(T delayMessage);}

AliceMessageHandler

packagecom.king.alice.rabbitmq;importcom.king.alice.common.json.JSON;importcom.king.alice.common.json.JSONObject;importcom.king.alice.rabbitmq.delay.bean.AliceMessage;importcom.king.alice.rabbitmq.delay.consumer.Strategy;importlombok.extern.slf4j.Slf4j;importorg.springframework.stereotype.Component;/**
 * @author 大魔王
 */@Component@Slf4jpublicclassAliceMessageHandlerimplementsStrategy<AliceMessage>{@Overridepublicvoidhandle(AliceMessage delayMessage){
        log.info("AliceMessage响应体{}",JSONObject.parseObject(JSON.toJSONString(delayMessage)));}}

UserMessageHandler

packagecom.king.alice.rabbitmq;importcom.king.alice.common.json.JSON;importcom.king.alice.common.json.JSONObject;importcom.king.alice.rabbitmq.delay.bean.UserMessage;importcom.king.alice.rabbitmq.delay.consumer.Strategy;importlombok.extern.slf4j.Slf4j;importorg.springframework.stereotype.Component;/**
 * @author 大魔王
 */@Slf4j@ComponentpublicclassUserMessageHandlerimplementsStrategy<UserMessage>{@Overridepublicvoidhandle(UserMessage delayMessage){
        log.info("UserMessage响应体{}",JSONObject.parseObject(JSON.toJSONString(delayMessage)));}}

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

SysAccountController

packagecom.king.alice.manage.sys.controller;importcn.hutool.core.date.DateUtil;importcom.king.alice.manage.sys.entity.SysAccount;importcom.king.alice.manage.sys.service.SysAccountService;importcom.king.alice.rabbitmq.delay.bean.AliceMessage;importcom.king.alice.rabbitmq.delay.bean.UserMessage;importcom.king.alice.rabbitmq.delay.provider.MessageProvider;importlombok.extern.slf4j.Slf4j;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.data.domain.Page;importorg.springframework.data.domain.PageRequest;importorg.springframework.http.ResponseEntity;importorg.springframework.web.bind.annotation.*;importjavax.annotation.Resource;importjava.util.Date;/**
 * 账号表(SysAccount)表控制层
 *
 * @author makejava
 * @since 2023-08-09 11:40:16
 */@RestController@Slf4jpublicclassSysAccountController{/**
     * 服务对象
     */@ResourceprivateSysAccountService sysAccountService;@AutowiredprivateMessageProvider messageProvider;@PostMapping("/send-alice-message")publicStringsendMsg(@RequestBodyAliceMessage aliceMessage){
        messageProvider.sendMessage(aliceMessage);
        log.info("当前时间：{},收到aliceMessage请求，msg:{}",DateUtil.format(newDate(),"yyyy-MM-dd HH:mm:ss"), aliceMessage);return"success";}@PostMapping("/send-user-message")publicStringsendMsg(@RequestBodyUserMessage userMessage){
        messageProvider.sendMessage(userMessage);
        log.info("当前时间：{},收到userMessage请求，msg:{}",DateUtil.format(newDate(),"yyyy-MM-dd HH:mm:ss"), userMessage);return"success";}}

调接口

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

标签： rabbitmq ruby 分布式
本文转载自: https://blog.csdn.net/TaoShao521/article/details/132305375
版权归原作者 大魔王的日常Log 所有， 如有侵权，请联系我们删除。
登录后发布评论

“一种多策略下RabbitMQ的延时队列实现”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？

文章导航