場景

用戶下單后，如果30min未支付，則刪除該訂單，這時候就要可以用延遲隊列

?

準備

利用rabbitmq_delayed_message_exchange插件；

首先下載該插件：https://www.rabbitmq.com/community-plugins.html

然后把該插件放到rabbitmq安裝目錄plugins下；

進入到sbin目錄下，執行"rabbitmq-plugins.bat enable rabbitmq_delayed_message_exchange";

?

關閉RabbitMQ服務，然后再啟動（直接重啟該插件可能會不生效）。

?

SpringBoot RabbitMQ代碼

application.properties配置文件

spring.application.name=spring-boot-rabbitmq spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=nut spring.rabbitmq.password=nut

?

配置類

注意這里的"x-delayed-type"和"x-delayed-message"

/*** 延遲隊列配置exchange*/ @Configuration public class DelayQueueConfig {public static final String DELAY_EXCHANGE = "DELAY_EXCHANGE";public static final String DELAY_QUEUE = "DELAY_QUEUE";public static final String DELAY_ROUTING_KEY = "DELAY_ROUTING_KEY";@Bean("delayExchange")public Exchange delayExchange() {Map<String, Object> args = new HashMap<>(1); // x-delayed-type 聲明 延遲隊列Exchange的類型args.put("x-delayed-type", "direct");return new CustomExchange(DELAY_EXCHANGE, "x-delayed-message",true, false,args);}@Bean("delayQueue")public Queue delayQueue() {return QueueBuilder.durable(DELAY_QUEUE).build();}/*** 將延遲隊列通過routingKey綁定到延遲交換器** @return*/@Beanpublic Binding delayQueueBindExchange() {return new Binding(DELAY_QUEUE, Binding.DestinationType.QUEUE, DELAY_EXCHANGE, DELAY_ROUTING_KEY, null);}}

?

生產者

發送消息時，指定延遲的毫秒

/*** 延遲隊列發送者*/ @Component @Slf4j public class DelayQueueSender {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendDelayQueue(int number) {log.warn("延遲隊列發送 : {} milliseconds", number);// 這里的Exchange可以是業務的Exchange，為了方便測試這里直接往死信Exchange里投遞消息 rabbitTemplate.convertAndSend(DelayQueueConfig.DELAY_EXCHANGE,DelayQueueConfig.DELAY_ROUTING_KEY,number, (message) -> {// 設置延遲的毫秒數message.getMessageProperties().setDelay(number);log.info("Now : {}", ZonedDateTime.now());return message;});} }

?

消費者

/*** 延遲隊列消費者*/ @Component @Slf4j @RabbitListener(queues = DelayQueueConfig.DELAY_QUEUE) public class DelayQueueConsumer {@RabbitHandlerpublic void receiveDelayMessage(Integer milliseconds){log.warn("DelayQueueConsumer Time : {}， and the millis : {}", ZonedDateTime.now(), milliseconds);}}

?

測試

先啟動項目；

然后在測試類中發送消息；

@RunWith(SpringRunner.class) @SpringBootTest public class RabbitmqApplicationTests {@Autowiredprivate DelayQueueSender delayQueueSender;@Testpublic void testDelayQueueSender(){delayQueueSender.sendDelayQueue(5000);} }

?

發送消息窗口：

?

消費者受到消息：

?

時間間隔證明延遲隊列發送完成！

?

參考：

https://blog.csdn.net/linsongbin1/article/details/80178122

https://blog.csdn.net/youjin/article/details/82586888

https://docs.spring.io/spring-amqp/docs/2.0.0.M2/reference/htmlsingle/#delayed-message-exchange

https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/

?

轉載于:https://www.cnblogs.com/theRhyme/p/10986409.html

總結

以上是生活随笔為你收集整理的SpringBoot RabbitMQ 延迟队列代码实现的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。