本文同步Java知音社區，專注于Java

作者：Sam哥哥http://blog.csdn.net/linsongbin1/article/details/80178122

概述

曾經去網易面試的時候，面試官問了我一個問題，說

下完訂單后，如果用戶未支付，需要取消訂單，可以怎么做

我當時的回答是，用定時任務掃描DB表即可。面試官不是很滿意，提出：

用定時任務無法做到準實時通知，有沒有其他辦法？

我當時的回答是：

可以用隊列，訂單下完后，發送一個消息到隊列里，并指定過期時間，時間一到，執行回調接口。

面試官聽完后，就不再問了。其實我當時的思路是對的，只不過講的不是很專業而已。專業說法是利用延遲消息。

其實用定時任務，確實有點問題，原本業務系統希望10分鐘后，如果訂單未支付，就馬上取消訂單，并釋放商品庫存。但是一旦數據量大的話，就會加長獲取未支付訂單數據的時間，部分訂單就做不到10分鐘后取消了，可能是15分鐘，20分鐘之類的。這樣的話，庫存就無法及時得到釋放，也就會影響成單數。而利用延遲消息，則理論上是可以做到按照設定的時間，進行訂單取消操作的。

目前網上關于使用RabbitMQ實現延遲消息的文章，大多都是講如何利用RabbitMQ的死信隊列來實現，實現方案看起來都很繁瑣復雜,并且還是使用原始的RabbitMQ Client API來實現的，更加顯得啰嗦。更多springboot整合實戰內容，可以在Java知音公眾號回復“springboot內容聚合”

Spring Boot 已經對RabbitMQ Client API進行了包裝，使用起來簡潔很多，下面詳細介紹一下如何利用rabbitmq_delayed_message_exchange 插件和Spring Boot來實現延遲消息。

軟件準備

erlang

請參考Win10下安裝erlang

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

本文使用的版本是:

• Erlang 20.3

RabbitMQ

請參考win10下安裝rabbitmq

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

本文使用的是window版本的RabbitMQ，版本號是:

• 3.7.4

rabbitmq_delayed_message_exchange插件

插件下載地址：

http://www.rabbitmq.com/community-plugins.html

打開網址后，ctrl + f，搜索rabbitmq_delayed_message_exchange。

千萬記住，一定選好版本號，由于我使用的是RabbitMQ 3.7.4,因此對應的rabbitmq_delayed_message_exchange插件也必須選擇3.7.x的。

如果沒有選對版本，在使用延遲消息的時候，會遇到各種各樣的奇葩問題，而且網上還找不到解決方案。我因為這個問題，折騰了整整一個晚上。請牢記，要選對插件版本。

下載完插件后，將其放置到RabbitMQ安裝目錄下的plugins目錄下，并使用如下命令啟動這個插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

如果啟動成功會出現如下信息：

The following plugins have been enabled: rabbitmq_delayed_message_exchange

啟動插件成功后，記得重啟一下RabbitMQ，讓其生效。

集成RabbitMQ

這個就非常簡單了,直接在maven工程的pom.xml文件中加入

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>

Spring Boot的版本我使用的是2.0.1.RELEASE.

接下來在application.properties文件中加入redis配置：

spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest

定義ConnectionFactory和RabbitTemplate

也很簡單，代碼如下：

package com.mq.rabbitmq;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;@Configuration @ConfigurationProperties(prefix = "spring.rabbitmq") public class RabbitMqConfig {private String host;private int port;private String userName;private String password;@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(host,port);cachingConnectionFactory.setUsername(userName);cachingConnectionFactory.setPassword(password);cachingConnectionFactory.setVirtualHost("/");cachingConnectionFactory.setPublisherConfirms(true);return cachingConnectionFactory;}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());return rabbitTemplate;}public String getHost() {return host;}public void setHost(String host) {this.host = host;}public int getPort() {return port;}public void setPort(int port) {this.port = port;}public String getUserName() {return userName;}public void setUserName(String userName) {this.userName = userName;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;} }

Exchange和Queue配置

package com.mq.rabbitmq;import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;import java.util.HashMap; import java.util.Map;@Configuration public class QueueConfig {@Beanpublic CustomExchange delayExchange() {Map<String, Object> args = new HashMap<>();args.put("x-delayed-type", "direct");return new CustomExchange("test_exchange", "x-delayed-message",true, false,args);}@Beanpublic Queue queue() {Queue queue = new Queue("test_queue_1", true);return queue;}@Beanpublic Binding binding() {return BindingBuilder.bind(queue()).to(delayExchange()).with("test_queue_1").noargs();} }

這里要特別注意的是，使用的是CustomExchange,不是DirectExchange，另外CustomExchange的類型必須是x-delayed-message。

實現消息發送

package com.mq.rabbitmq;import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service;import java.text.SimpleDateFormat; import java.util.Date;@Service public class MessageServiceImpl {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMsg(String queueName,String msg) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("消息發送時間:"+sdf.format(new Date()));rabbitTemplate.convertAndSend("test_exchange", queueName, msg, new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {message.getMessageProperties().setHeader("x-delay",3000);return message;}});} }

注意在發送的時候，必須加上一個header

• x-delay

在這里我設置的延遲時間是3秒。

消息消費者

package com.mq.rabbitmq;import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component;import java.text.SimpleDateFormat; import java.util.Date;@Component public class MessageReceiver {@RabbitListener(queues = "test_queue_1")public void receive(String msg) {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("消息接收時間:"+sdf.format(new Date()));System.out.println("接收到的消息:"+msg);} }

運行Spring Boot程序和發送消息

直接在main方法里運行Spring Boot程序,Spring Boot會自動解析MessageReceiver類的。

接下來只需要用Junit運行一下發送消息的接口即可。

package com.mq.rabbitmq;import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner;@RunWith(SpringRunner.class) @SpringBootTest public class RabbitmqApplicationTests {@Autowiredprivate MessageServiceImpl messageService;@Testpublic void send() {messageService.sendMsg("test_queue_1","hello i am delay msg");}}

運行完后，可以看到如下信息：

消息發送時間:2018-05-03 12:44:53

3秒鐘后，Spring Boot控制臺會輸出:

消息接收時間:2018-05-03 12:44:56
接收到的消息:hello i am delay msg

總結

以上是生活随笔為你收集整理的spring boot rabbitmq_Spring Boot+RabbitMQ 实现延迟消息实现完整版，实用！的全部內容，希望文章能夠幫你解決所遇到的問題。

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