RabbitMQ

安装RabitMQ

1：安装RabbitMQ需要先安装Erlang语言开发包。下载地址 http://www.erlang.org/download.html 在win7下安装Erlang最好默认安装。
配置环境变量 ERLANG_HOME C:\Program Files (x86)\erl5.9
添加到PATH %ERLANG_HOME%\bin;
2：安装RabbitMQ 下载地址 http://www.rabbitmq.com/download.html 安装教程：http://www.rabbitmq.com/install-windows.html
配置环境变量 C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-2.8.0
添加到PATH %RABBITMQ_SERVER%\sbin;
3：进入%RABBITMQ_SERVER%\sbin 目录以管理员身份运行 rabbitmq-plugins.bat
rabbitmq-plugins.bat enable rabbitmq_management
安装完成之后以管理员身份启动 rabbitmq-service.bat
rabbitmq-service.bat stop
rabbitmq-service.bat install
rabbitmq-service.bat start
4：浏览器访问localhost:15672 默认账号：guest 密码：guest

Java 开发RabitMQ

1.下载jar

http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.5.4/rabbitmq-java-client-bin-3.5.4.zip

2.生产者

1package com.rabbit;
2
3
4import com.rabbitmq.client.*;
5
6public class Send {
7
8    private final static String QUEUE_NAME = "hello";  
9    
10    public static void main(String[] args) throws Exception {  
11        ConnectionFactory factory = new ConnectionFactory();  
12        factory.setHost("localhost");  
13        Connection connection = factory.newConnection();  
14        Channel channel = connection.createChannel();  
15        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
16        String message = "Hello World!";  
17        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());  
18        System.out.println(" [x] Sent '" + message + "'");  
19        channel.close();  
20        connection.close();  
21    }  
22}

3.消费者

1package com.rabbit;
2
3import com.rabbitmq.client.Channel;  
4import com.rabbitmq.client.Connection;  
5import com.rabbitmq.client.ConnectionFactory;  
6import com.rabbitmq.client.QueueingConsumer;  
7  
8public class Reqv {  
9    private final static String QUEUE_NAME = "hello";  
10  
11    public static void main(String[] argv) throws Exception {  
12  
13        ConnectionFactory factory = new ConnectionFactory();  
14        factory.setHost("localhost");  
15        Connection connection = factory.newConnection();  
16        Channel channel = connection.createChannel();  
17  
18        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
19        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");  
20  
21        QueueingConsumer consumer = new QueueingConsumer(channel);  
22        channel.basicConsume(QUEUE_NAME, true, consumer);  
23  
24        while (true) {  
25            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
26            String message = new String(delivery.getBody());  
27            System.out.println(" [x] Received '" + message + "'");  
28        }  
29    }  
30}

Spring RabitMQ

先看一个帖子

1.首先是生产者配置

<? 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: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.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context.xsd

http://www.springframework.org/schema/rabbit

http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">

< rabbit:connection-factory id = "connectionFactory" host = "localhost" username = "guest"

password = "guest" port = "5672" />

< rabbit:admin connection-factory = "connectionFactory" />

< rabbit:queue id = "queue_one" durable = "true" auto-delete = "false" exclusive = "false" name = "queue_one" />

< rabbit:direct-exchange name = "my-mq-exchange" durable = "true" auto-delete = "false" id = "my-mq-exchange" >

< rabbit:bindings >

< rabbit:binding queue = "queue_one" key = "queue_one_key" />

</ rabbit:bindings >

</ rabbit:direct-exchange >

<-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->

< bean id = "jsonMessageConverter" class = "mq.convert.FastJsonMessageConverter" ></ bean >

<-- spring template声明-->

< rabbit:template exchange = "my-mq-exchange" id = "amqpTemplate" connection-factory = "connectionFactory" message-converter = "jsonMessageConverter" />

</ beans >

2.fastjson messageconver插件实现

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.MessageProperties;

import org.springframework.amqp.support.converter.AbstractMessageConverter;

import org.springframework.amqp.support.converter.MessageConversionException;

import fe.json.FastJson;

public class FastJsonMessageConverter extends AbstractMessageConverter {

private static Log log = LogFactory.getLog(FastJsonMessageConverter. class );

public static final String DEFAULT_CHARSET = "UTF-8" ;

private volatile String defaultCharset = DEFAULT_CHARSET;

public FastJsonMessageConverter() {

super ();

//init();

}

public void setDefaultCharset(String defaultCharset) {

this .defaultCharset = (defaultCharset != null ) ? defaultCharset

: DEFAULT_CHARSET;

}

public Object fromMessage(Message message)

throws MessageConversionException {

return null ;

}

public <T> T fromMessage(Message message,T t) {

String json = "" ;

try {

json = new String(message.getBody(),defaultCharset);

} catch (UnsupportedEncodingException e) {

e.printStackTrace();

}

return (T) FastJson.fromJson(json, t.getClass());

}

protected Message createMessage(Object objectToConvert,

MessageProperties messageProperties)

throws MessageConversionException {

byte [] bytes = null ;

try {

String jsonString = FastJson.toJson(objectToConvert);

bytes = jsonString.getBytes( this .defaultCharset);

} catch (UnsupportedEncodingException e) {

throw new MessageConversionException(

"Failed to convert Message content" , e);

}

messageProperties.setContentType(MessageProperties.CONTENT_TYPE_JSON);

messageProperties.setContentEncoding( this .defaultCharset);

if (bytes != null ) {

messageProperties.setContentLength(bytes.length);

}

return new Message(bytes, messageProperties);

}

3.生产者端调用

import java.util.List;

import org.springframework.amqp.core.AmqpTemplate;

public class MyMqGatway {

@Autowired

private AmqpTemplate amqpTemplate;

public void sendDataToCrQueue(Object obj) {

amqpTemplate.convertAndSend( "queue_one_key" , obj);

}

4.消费者端配置(与生产者端大同小异)

<? 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: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.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context.xsd

http://www.springframework.org/schema/rabbit

http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">

< rabbit:connection-factory id = "connectionFactory" host = "localhost" username = "guest"

password = "guest" port = "5672" />

< rabbit:admin connection-factory = "connectionFactory" />

< rabbit:queue id = "queue_one" durable = "true" auto-delete = "false" exclusive = "false" name = "queue_one" />

< rabbit:direct-exchange name = "my-mq-exchange" durable = "true" auto-delete = "false" id = "my-mq-exchange" >

< rabbit:bindings >

< rabbit:binding queue = "queue_one" key = "queue_one_key" />

</ rabbit:bindings >

</ rabbit:direct-exchange >

< rabbit:listener-container connection-factory = "connectionFactory" acknowledge = "auto" task-executor = "taskExecutor" >

< rabbit:listener queues = "queue_one" ref = "queueOneLitener" />

</ rabbit:listener-container >

</ beans >

5.消费者端调用

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.MessageListener;

public class QueueOneLitener implements MessageListener{

@Override

public void onMessage(Message message) {

System.out.println( " data :" + message.getBody());

}

6.由于消费端当队列有数据到达时，对应监听的对象就会被通知到，无法做到批量获取，批量入库，因此可以在消费端缓存一个临时队列，将mq取出来的数据存入本地队列，后台线程定时批量处理即可

网上的这篇帖子,将路由设置成了direct,这样子保证了一对一的读取和发布,对于路由的exchange请参照这里的说明:http://melin.iteye.com/blog/691265

要特别注意的是,queue队列,其实只是在xml中的配置,并没有实际的意义,读取的时候使用的键值其实就是key,在发送的时候 conv``ertAndSend(``"queue_one_key"``, obj),就是键值,不要混淆了

下面介绍我在项目中的应用,这里还可以实现如果有异常,实现了自动重发功能

1、生产者

app.xml配置,将其引入到spring.xml中即可

1<?xml version="1.0" encoding="UTF-8"?>  
2<beans xmlns="http://www.springframework.org/schema/beans"  
3	xmlns:aop="http://www.springframework.org/schema/aop"   
4	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"   
5	xmlns:context="http://www.springframework.org/schema/context"
6	xmlns:rabbit="http://www.springframework.org/schema/rabbit"
7	xsi:schemaLocation="http://www.springframework.org/schema/beans   
8	http://www.springframework.org/schema/beans/spring-beans-3.1.xsd    
9	http://www.springframework.org/schema/rabbit 
10	http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
11
12	<!-- 连接服务配置  -->
13	<rabbit:connection-factory id="connectionFactory" 
14		host="${rabbitmq.host}" port="${rabbitmq.port}" 
15		username="${rabbitmq.username}" password="${rabbitmq.password}"/>  
16  	
17  	<!-- 创建rabbitAdmin 代理类 -->
18    <rabbit:admin connection-factory="connectionFactory" />  
19    
20    <!-- 创建rabbitTemplate 消息模板类 --> 
21    <rabbit:template id="rabbitTemplate"  connection-factory="connectionFactory"/>
22    
23</beans>

发送数据

1package com.zefun.web.service;
2import java.util.HashMap;
3import java.util.Map;
4
5import net.sf.json.JSONObject;
6
7import org.apache.log4j.Logger;
8import org.springframework.amqp.rabbit.core.RabbitTemplate;
9import org.springframework.beans.factory.annotation.Autowired;
10import org.springframework.beans.factory.annotation.Qualifier;
11import org.springframework.stereotype.Service;
12
13import com.zefun.common.consts.App;
14
15/**
16 * 消息队列服务类
17* @author 
18* @date Aug 24, 2015 3:51:04 PM 
19*/
20@Service
21public class RabbitService {
22    /** 日志对象 */
23    private static Logger logger = Logger.getLogger(RabbitService.class);
24
25    /** rabbit队列模版方法 */
26    @Autowired()
27    private RabbitTemplate rabbitTemplate;
28    
29    /**
30     * 发送优惠券队列
31    * @author 高国藩
32    * @date 2015年9月16日 上午11:34:12
33    * @param map 参数
34     */
35    public void sendCoupons(Map<String, Object> map) {
36        //App.Queue.SEND_COUPONS 就是对应key值,在消费的时候使用该值
37        rabbitTemplate.convertAndSend(App.Queue.SEND_COUPONS, map);
38    }
39}

1、消费者

app.xml

1<?xml version="1.0" encoding="UTF-8"?>
2<beans xmlns="http://www.springframework.org/schema/beans"
3    xmlns:aop="http://www.springframework.org/schema/aop" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4    xmlns:context="http://www.springframework.org/schema/context"
5    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
6    xsi:schemaLocation="http://www.springframework.org/schema/beans
7    http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
8    http://www.springframework.org/schema/rabbit
9    http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd
10    http://www.springframework.org/schema/aop
11    http://www.springframework.org/schema/aop/spring-aop-3.1.xsd
12    http://www.springframework.org/schema/context
13    http://www.springframework.org/schema/context/spring-context-3.1.xsd">
14
15    <!-- 连接服务配置 -->
16    <rabbit:connection-factory id="connectionFactory"
17        host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}"
18        password="${rabbitmq.password}" channel-cache-size="${rabbitmq.channel.cache.size}" />
19
20    <!-- 创建rabbitAdmin 代理类 -->
21    <rabbit:admin connection-factory="connectionFactory" />
22    
23   <rabbit:queue id="queue_member_service_coupon" name="${rabbitmq.wechat.template.notice.coupon}" durable="true"
24        auto-delete="false" exclusive="false" />
25
26    <!--路由设置 将队列绑定，属于direct类型 -->
27    <rabbit:direct-exchange id="directExchange"
28        name="directExchange" durable="true" auto-delete="false">
29        <rabbit:bindings>
30            <rabbit:binding queue="queue_member_service_coupon" key="${rabbitmq.wechat.template.notice.coupon}" />
31        </rabbit:bindings>
32    </rabbit:direct-exchange>
33    
34    <bean id="ackManual"
35        class="org.springframework.beans.factory.config.FieldRetrievingFactoryBean">
36        <property name="staticField"
37            value="org.springframework.amqp.core.AcknowledgeMode.MANUAL" />
38    </bean>
39
40
41    <!-- 优惠券发送通知 -->
42    <bean
43        class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer">
44        <property name="connectionFactory" ref="connectionFactory" />
45        <property name="acknowledgeMode" ref="ackManual" />
46        <property name="queueNames" value="${rabbitmq.wechat.template.notice.coupon}" />
47        <property name="messageListener">
48            <bean class="com.zefun.wechat.listener.MemberTranscationNoitceCoupon" />
49        </property>
50        <property name="concurrentConsumers" value="${rabbitmq.concurrentConsumers}" />
51        <property name="adviceChain">
52            <bean
53                class="org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptorFactoryBean">
54                <property name="messageRecoverer">
55                    <bean class="com.zefun.wechat.utils.MQRepublishMessageRecoverer"/>
56                </property>
57                <property name="retryOperations">
58                    <bean class="org.springframework.retry.support.RetryTemplate">
59                        <property name="backOffPolicy">
60                            <bean
61                                class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
62                                <property name="initialInterval" value="500" />
63                                <property name="multiplier" value="10.0" />
64                                <property name="maxInterval" value="10000" />
65                            </bean>
66                        </property>
67                    </bean>
68                </property>
69            </bean>
70        </property>
71        <property name="errorHandler">
72            <bean class="com.zefun.wechat.utils.MQErrorHandler"/>
73        </property>
74    </bean>
75    
76    <bean id="msgConverter" class="org.springframework.amqp.support.converter.SimpleMessageConverter" />
77
78    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
79    
80</beans>

消费类

1package com.zefun.wechat.listener;
2
3import java.util.Map;
4
5import net.sf.json.JSONObject;
6
7import org.apache.log4j.Logger;
8import org.springframework.amqp.core.Message;
9import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
10import org.springframework.amqp.support.converter.MessageConversionException;
11import org.springframework.amqp.support.converter.MessageConverter;
12import org.springframework.beans.factory.annotation.Autowired;
13
14import com.rabbitmq.client.Channel;
15import com.zefun.wechat.service.RedisService;
16import com.zefun.wechat.utils.App;
17import com.zefun.wechat.utils.HttpClientUtil;
18
19
20public class MemberTranscationNoitceCoupon implements ChannelAwareMessageListener{
21    
22    @Autowired
23    private MessageConverter msgConverter;
24    @Autowired
25    private RedisService redisService;
26
27    private static final Logger logger = Logger.getLogger(EmployeeServiceNoticeListener.class);
28    
29    @Override
30    public void onMessage(Message message, Channel channel) throws Exception {
31
32        Object obj = null;
33        try {
34            obj = msgConverter.fromMessage(message);
35        } catch (MessageConversionException e) {
36            logger.error("convert MQ message error.", e);
37        } finally {
38            long deliveryTag = message.getMessageProperties().getDeliveryTag();
39            if (deliveryTag != App.DELIVERIED_TAG) {
40                channel.basicAck(deliveryTag, false);
41                message.getMessageProperties().setDeliveryTag(App.DELIVERIED_TAG);
42                logger.info("revice and ack msg: " + (obj == null ? message : new String((byte[]) obj)));
43            }
44        }
45        if (obj == null) {
46            return;
47        }
48        Map<?, ?> map = (Map<?, ?>) obj;
49        HttpClientUtil.sendPost(getTemplSendUrl(map.get("storeId").toString()), JSONObject.fromObject(map).toString(), null);
50        boolean flag = false;
51        if (!flag) {
52            logger.error("hanler message " + obj + " failed, throw a exception, and it will be retried.");
53            throw new RuntimeException("hanler message " + obj + " failed.");//如果此处抛出了异常,那么在消息转换类中会接受到,触发重新加入队列中的时间
54        }
55    }
56
57    private String getTemplSendUrl(String storeId) {
58        String accessToken = redisService.hget(App.Redis.STORE_WECHAT_ACCESS_TOKEN_KEY_HASH, storeId);
59        return "https://api.weixin.qq.com/cgi-bin/message/template/send?access_token=" + "HwqNKfoP287p4ddayRVX0PN1-8GFukq776MmwQqaL4OI2oEb4WGzclCaPgQIZZd4I42Xo-beX4XrW5Og3NblI_Auf5dGj1hdPrBuhz5OYHE";
60    }
61}

消息转换类

1package com.zefun.wechat.utils;
2
3import java.io.PrintWriter;
4import java.io.StringWriter;
5import java.util.Map;
6
7import org.apache.log4j.Logger;
8import org.springframework.amqp.core.Message;
9import org.springframework.amqp.rabbit.core.RabbitTemplate;
10import org.springframework.amqp.rabbit.retry.MessageRecoverer;
11import org.springframework.amqp.support.converter.MessageConverter;
12import org.springframework.beans.factory.annotation.Autowired;
13
14public class MQRepublishMessageRecoverer implements MessageRecoverer {
15	
16	private static final Logger logger = Logger.getLogger(MQRepublishMessageRecoverer.class);
17
18	@Autowired
19	private RabbitTemplate rabbitTemplate;
20	
21	@Autowired
22	private MessageConverter msgConverter;
23
24	@Override
25	public void recover(Message message, Throwable cause) {
26		Map<String, Object> headers = message.getMessageProperties().getHeaders();
27		headers.put("x-exception-stacktrace", getStackTraceAsString(cause));
28		headers.put("x-exception-message", cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage());
29		headers.put("x-original-exchange", message.getMessageProperties().getReceivedExchange());
30		headers.put("x-original-routingKey", message.getMessageProperties().getReceivedRoutingKey());
31		//重新将数据放回队列中
32		this.rabbitTemplate.send(message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), message);
33		logger.error("handler msg (" + msgConverter.fromMessage(message) + ") err, republish to mq.", cause);
34	}
35
36	private String getStackTraceAsString(Throwable cause) {
37		StringWriter stringWriter = new StringWriter();
38		PrintWriter printWriter = new PrintWriter(stringWriter, true);
39		cause.printStackTrace(printWriter);
40		return stringWriter.getBuffer().toString();
41	}
42}
43
44
45
46package com.zefun.wechat.utils;
47
48import java.lang.reflect.Field;
49import java.util.Date;
50
51import org.apache.commons.lang.reflect.FieldUtils;
52import org.apache.log4j.Logger;
53import org.springframework.amqp.core.Message;
54import org.springframework.amqp.support.converter.MessageConverter;
55import org.springframework.beans.factory.annotation.Autowired;
56import org.springframework.util.ErrorHandler;
57
58import com.zefun.wechat.service.RedisService;
59
60public class MQErrorHandler implements ErrorHandler {
61
62    private static final Logger logger = Logger.getLogger(MQErrorHandler.class);
63
64    @Autowired
65    private RedisService redisService;
66    @Autowired
67    private MessageConverter msgConverter;
68
69    @Override
70    public void handleError(Throwable cause) {
71        Field mqMsgField = FieldUtils.getField(MQListenerExecutionFailedException.class, "mqMsg", true);
72        if (mqMsgField != null) {
73            try {
74                Message mqMsg = (Message) mqMsgField.get(cause);
75                Object msgObj = msgConverter.fromMessage(mqMsg);
76                logger.error("handle MQ msg: " + msgObj + " failed, record it to redis.", cause);
77                redisService.zadd(App.MsgErr.MQ_MSG_ERR_RECORD_KEY, new Double(new Date().getTime()), msgObj.toString());
78            } catch (Exception e) {
79                e.printStackTrace();
80            }
81        } else {
82            logger.error("An error occurred.", cause);
83        }
84    }
85
86}
87
88
89
90package com.zefun.wechat.utils;
91
92import org.springframework.amqp.core.Message;
93import org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException;
94
95public class MQListenerExecutionFailedException extends
96        ListenerExecutionFailedException {
97
98    private static final long serialVersionUID = 1L;
99
100    private Message mqMsg;
101
102    public MQListenerExecutionFailedException(String msg, Throwable cause) {
103        super(msg, cause);
104    }
105
106    public MQListenerExecutionFailedException(String msg, Message mqMsg, Throwable cause) {
107        this(msg, cause);
108        this.mqMsg = mqMsg;
109    }
110
111    public Message getMqMsg() {
112        return mqMsg;
113    }
114
115}
116
117
118
119package com.zefun.wechat.utils;
120
121import org.springframework.amqp.AmqpRejectAndDontRequeueException;
122import org.springframework.amqp.core.Message;
123import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer;
124
125public class MQRejectAndDontRequeueRecoverer extends
126        RejectAndDontRequeueRecoverer {
127
128    @Override
129    public void recover(Message message, Throwable cause) {
130        throw new MQListenerExecutionFailedException("Retry Policy Exhausted", message,
131                new AmqpRejectAndDontRequeueException(cause));
132    }
133
134}

properties内容

1rabbitmq.host=127.0.0.1
2rabbitmq.port=5672
3rabbitmq.username=guest
4rabbitmq.password=guest
5rabbitmq.concurrentConsumers=5
6rabbitmq.channel.cache.size=50
7rabbitmq.wechat.template.notice.coupon=queue_member_service_coupon

maven

1<!-- springmvc rabbitmq -->
2		<dependency>
3			<groupId>com.rabbitmq</groupId>
4			<artifactId>amqp-client</artifactId>
5			<version>3.3.0</version>
6		</dependency>
7		<dependency>
8			<groupId>org.springframework.amqp</groupId>
9			<artifactId>spring-amqp</artifactId>
10			<version>1.3.0.RELEASE</version>
11		</dependency>
12
13		<dependency>
14			<groupId>org.springframework.amqp</groupId>
15			<artifactId>spring-rabbit</artifactId>
16			<version>1.3.0.RELEASE</version>
17		</dependency>
18
19		<dependency>
20			<groupId>org.springframework.retry</groupId>
21			<artifactId>spring-retry</artifactId>
22			<version>1.0.3.RELEASE</version>
23		</dependency>

本案例下载

http://download.csdn.net/detail/u014201191/9113775

http://download.csdn.net/detail/u014201191/9341163

RabbitMQ java Spring 开发教程

安装RabitMQ

Java 开发RabitMQ

Spring RabitMQ

本案例下载

评论区

相关推荐