杨幂刘恺威结婚视频录像,杨幂小视频

Spring集成RabbitMQ并實(shí)現(xiàn)延遲隊(duì)列

2018.02.26

一、說(shuō)明

在實(shí)際業(yè)務(wù)場(chǎng)景中可能會(huì)用到延時(shí)消息發(fā)送，例如異步回調(diào)失敗時(shí)的重發(fā)機(jī)制。 RabbitMQ本身不具有延時(shí)消息隊(duì)列的功能，但是可以通過(guò)rabbitmq-delayed-message-exchange來(lái)實(shí)現(xiàn)（也可以通過(guò)TTL(Time To Live)、DLX(Dead Letter Exchanges)特性實(shí)現(xiàn)，我們主要講解通過(guò)延遲插件來(lái)實(shí)現(xiàn)的方法）。利用RabbitMQ的這種特性，應(yīng)該可以實(shí)現(xiàn)很多現(xiàn)實(shí)中的業(yè)務(wù)，我們可以發(fā)揮想象。

二、安裝插件

RabbitMQ的安裝請(qǐng)參考我的文章“RabbitMQ安裝與使用”，這里我們重點(diǎn)講插件的安裝。

首先到http://www.rabbitmq.com/community-plugins.html網(wǎng)頁(yè)下載適合的“rabbitmq_delayed_message_exchange插件”。下載完成后將它放到RabbitMQ插件安裝目錄（{rabbitmq-server}/plugins/），然后執(zhí)行命令rabbitmq-plugins enable rabbitmq_delayed_message_exchange啟用插件，執(zhí)行命令rabbitmq-plugins disable rabbitmq_delayed_message_exchange也可以關(guān)閉插件。具體過(guò)程可以查看參考文檔2。

三、Spring集成RabbitMQ

1、maven配置

[html] view plain copy

print?

<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-amqp</artifactId>
<version>1.6.6.RELEASE</version>
<exclusions>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>4.1.6.RELEASE</version>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>1.6.6.RELEASE</version>
<exclusions>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-core</artifactId>
<version>4.1.6.RELEASE</version>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-messaging</artifactId>
<version>4.1.6.RELEASE</version>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
<version>4.1.6.RELEASE</version>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>4.1.6.RELEASE</version>
</exclusion>
</exclusions>
</dependency>

說(shuō)明：實(shí)現(xiàn)延遲隊(duì)列需要Spring在4.1以上，spring-amqp在1.6以上。

2、xml配置

[html] view plain copy

print?

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:util="http://www.springframework.org/schema/util" xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.1.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.1.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd">
<context:property-placeholder location="classpath:rmq-config.properties" ignore-unresolvable="true"/>
<bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
<property name="host" value="${rabbitmq.host}" />
<property name="port" value="${rabbitmq.port}" />
<property name="username" value="${rabbitmq.username}" />
<property name="password" value="${rabbitmq.password}" />
<property name="channelCacheSize" value="${rabbitmq.channel.cacheSize}" />
</bean>
<bean id="orderConsumer" class="com.xxx.rmq.OrderConsumer"></bean>
<bean id="messageConverter" class="org.springframework.amqp.support.converter.SimpleMessageConverter" />
<bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
<rabbit:admin connection-factory="connectionFactory" />
<rabbit:topic-exchange name="delay_exchange" delayed="true">
<rabbit:bindings>
<rabbit:binding queue="delay_queue" pattern="order.delay.notify" />
</rabbit:bindings>
</rabbit:topic-exchange>
<rabbit:queue name="delay_queue" durable="true" auto-declare="true" auto-delete="false" />
<rabbit:template id="delayMsgTemplate" connection-factory="connectionFactory" message-converter="jsonMessageConverter" exchange="delay_exchange" />
<rabbit:listener-container connection-factory="connectionFactory" channel-transacted="false" acknowledge="auto" message-converter="jsonMessageConverter">
<rabbit:listener queues="delay_queue" ref="orderConsumer" method="delayMsg" />
</rabbit:listener-container>
</beans>

說(shuō)明：spring-rabbit-1.6.xsd必須是1.6及以上版本，否則會(huì)報(bào)“元素 'rabbit:topic-exchange' 中不允許出現(xiàn)屬性 'delayed'”錯(cuò)誤。具體請(qǐng)查看參考文檔3。

四、延遲隊(duì)列的使用

1、發(fā)送消息Producer

[java] view plain copy

print?

import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
*
* @author Horace
* @version 創(chuàng)建時(shí)間：2016年10月26日下午6:34:31
*/
@Service
public class MessageProducerServiceImpl implements MessageProducerService{
@Autowired
private AmqpTemplate delayMsgTemplate;
@Override
public void delayMsg(JSONObject msg,int delay) {
// TODO Auto-generated method stub
final int xdelay= delay*1000;
delayMsgTemplate.convertAndSend("order.delay.notify", (Object) msg,
new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message)
throws AmqpException {
// TODO Auto-generated method stub
message.getMessageProperties().setDelay(xdelay);
return message;
}
});
}
}

2、異步接收消息Consumer

[java] view plain copy

print?

import net.sf.json.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
/**
*
* @author Horace
* @version 創(chuàng)建時(shí)間：2016年10月26日下午2:48:14
*/
public class OrderConsumer {
private static Logger logger = LoggerFactory.getLogger(OrderConsumer.class);
@Autowired
private MessageProducerService messageProducerService;
public void delayMsg(Object obj) {
logger.info("[延時(shí)消息]" + obj);
if (obj != null) {
JSONObject notifyJson = JSONObject.fromObject(obj);
String notifyUrl = notifyJson.getString("notifyUrl");
String notifyContent = notifyJson.getString("notifyContent");
String result = HttpUtil.postMessage(notifyUrl, notifyContent);
if (StringUtils.isBlank(result)) { // 通知失敗進(jìn)入重發(fā)機(jī)制
int newNotifyCount = notifyJson.getInt("notifyCount") + 1; //已經(jīng)通知的次數(shù)
if (newNotifyCount < 5) {
notifyJson.put("notifyCount", newNotifyCount);
int spacingInterval = getSpacingInterval(newNotifyCount);
messageProducerService
.delayMsg(notifyJson, spacingInterval);
} else {
logger.info("通知5次都失敗，等待后臺(tái)手工處理！");
}
}
}
}
/**
* 重復(fù)通知間隔時(shí)間（單位為秒）
* @param notifyCount 已經(jīng)通知的次數(shù)
* @return
*/
private int getSpacingInterval(int notifyCount) {
// TODO Auto-generated method stub
int spacingInterval = 0;
switch (notifyCount) {
case 1:
spacingInterval = 10;
break;
case 2:
spacingInterval = 20;
break;
case 3:
spacingInterval = 30;
break;
case 4:
spacingInterval = 60;
break;
case 5:
spacingInterval = 90;
break;
default:
break;
}
return spacingInterval;
}
}

參考文檔：

1、http://blog.csdn.net/tongdao/article/details/51638066 RabbitMQ安裝與使用

2、http://blog.csdn.net/u014308482/article/details/53036770 rabbitmq 實(shí)現(xiàn)延遲隊(duì)列的兩種方式

3、http://docs.spring.io/spring-amqp/docs/1.6.0.RELEASE/reference/html/_reference.html#delayed-message-exchange

本站僅提供存儲(chǔ)服務(wù)，所有內(nèi)容均由用戶發(fā)布，如發(fā)現(xiàn)有害或侵權(quán)內(nèi)容，請(qǐng)點(diǎn)擊舉報(bào)。

打開(kāi)APP，閱讀全文并永久保存查看更多類似文章

springboot整合rabbitMQ

11-RabbitMQ高級(jí)特性-消息可靠性投遞

RabbitMQ與java、Spring結(jié)合實(shí)例詳細(xì)講解

pom.xml

RabbitMQ教程

SpringMVC和rabbitmq集成的使用案例

更多類似文章 >>

国产一级a片免费看高清,亚洲熟女中文字幕在线视频,黄三级高清在线播放,免费黄色视频在线看