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";}
}