![51%71ZMOPR]O154T3CNI$@1.png](http://upload-images.jianshu.io/upload_images/3269064-2b5de5b351294297.png?imageMogr2/auto-orient/strip%7CimageView2/2/w/1240)
1. ActiveMQ安裝
1.1 下載(版本5.14.5)
1.2 安裝
解壓下載的壓縮文件到任意目錄中(
eg. C:\Program Files (x86)\apache-activemq-5.14.5)革为，進(jìn)入%ACTIVEMQ_HOME%/bin目錄匙奴，根據(jù)自己的系統(tǒng)位數(shù)小压，進(jìn)入32/64目錄艇潭，點擊activemq.bat啟動ActiveMQ囤攀；
2. ActiveMQ與Spring整合使用
2.1 在Maven中添加ActiveMQ和JMS相關(guān)的pom，如下：
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>4.2.5.RELEASE</version>
<!--<version>{spring.version}</version>-->
</dependency>
<!-- xbean 如<amq:connectionFactory /> -->
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-core</artifactId>
<version>5.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.12.1</version>
</dependency>
2.2 添加配置文件spring-activemq.xml
在配置文件中加入以下配置信息裁赠，每個配置信息都有具體的解釋：
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.apache.org/schema/core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">
<!--配置連接ActiveMQ的連接基本信息 -->
<amq:connectionFactory id="amqConnectionFactory"
brokerURL="tcp://localhost:61616" userName="admin" password="admin" />
<!-- 配置JMS連接工廠 -->
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="amqConnectionFactory" />
<property name="sessionCacheSize" value="100" />
</bean>
<!-- 定義消息隊列（Queue） -->
<bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 設(shè)置消息隊列的名字 -->
<constructor-arg>
<value>testQueue</value>
</constructor-arg>
</bean>
<!-- 配置JMS模板（Queue）问窃，Spring提供的JMS工具類，它發(fā)送傻寂、接收消息息尺。 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="demoQueueDestination" />
<property name="receiveTimeout" value="10000" />
<!-- true是topic，false是queue疾掰，默認(rèn)是false -->
<property name="pubSubDomain" value="false" />
</bean>
<!-- 配置消息隊列監(jiān)聽者（Queue） -->
<!-- 打開監(jiān)聽器搂誉，會立即去消費消息（即，起到實時消費通信的作用） -->
<!-- <bean id="queueMessageListener" class="com.hp.common.listener.QueueMessageListener"></bean> -->
<!-- 顯示注入消息監(jiān)聽容器（Queue）静檬，配置連接工廠炭懊，監(jiān)聽的目標(biāo)是demoQueueDestination，監(jiān)聽器是上面定義的監(jiān)聽器 -->
<!-- <bean id="queueListenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="demoQueueDestination" />
<property name="messageListener" ref="queueMessageListener" />
</bean>
-->
</beans>
注：在配置文件中拂檩，一定不要忘記加入ActiveMQ和JMS相關(guān)的schema
2.3 創(chuàng)建Producer和Consumer相關(guān)的Service
創(chuàng)建ProducerService侮腹，用于發(fā)送信息到消息中心
@Service
public class ProducerService {
@Resource(name = "jmsTemplate")
private JmsTemplate jmsTemplate;
private Queue queue;
/**
* 根據(jù)目的地發(fā)送消息
*/
public void sendMessage(Destination destination, final String msg) {
System.out.println(Thread.currentThread().getName() + " 向隊列" + destination.toString()
+ "發(fā)送消息------->" + msg);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
public String send(String userId, String msg) {
System.out.println(
Thread.currentThread().getName() + " 向 " + userId + " 的隊列" + userId.toString() + "發(fā)送消息------>" + msg);
queue = new ActiveMQQueue(userId);
jmsTemplate.send(queue, new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message=session.createTextMessage(msg);
message.setStringProperty(userId, msg);
return message;
}
});
return "發(fā)送成功";
}
/**
* 向默認(rèn)目的地發(fā)送消息
*/
public String sendMessage(final String msg) {
String destination = jmsTemplate.getDefaultDestinationName();
System.out
.println(Thread.currentThread().getName() + " 向隊列" + destination + "發(fā)送消息---------------------->" + msg);
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
return "發(fā)送成功";
}
}
創(chuàng)建ConsumerService，用于接受消息
@Service
public class ConsumerService{
@Resource(name = "jmsTemplate")
private JmsTemplate jmsTemplate;
public String receive(Destination destination) {
TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
try {
System.out.println("從隊列" + destination.toString() + "收到了消息：\t" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
return textMessage.toString();
}
public String receive(String userId) {
Queue queue=new ActiveMQQueue(userId+"?consumer.prefetchSize=4");
Message message = null;
String property=null;
try {
message=jmsTemplate.receive(queue);
property=message.getStringProperty(userId);
System.out.println("從隊列" + queue.toString() + "收到了消息：\t" + property);
} catch (JMSException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
return property;
}
}
2.4 添加Controller稻励，用于曝露接口
@Controller
@RequestMapping(value="/mq")
public class MessageController {
private Logger logger = Logger.getLogger(MessageController.class);
@Resource(name = "demoQueueDestination")
private Destination destination;
@Autowired
private ProducerService producer;
@Autowired
private ConsumerService consumer;
@RequestMapping(value = "/SendMessage", method = RequestMethod.POST,produces="application/json")
@ResponseBody
public void send(@RequestParam(value = "userId",required=false)String userId,@RequestParam(value = "msg")String msg) {
logger.info(Thread.currentThread().getName() + "------------send to jms Start");
if (userId==null||"".equals(userId)) {
producer.sendMessage(destination, msg);
}else {
producer.send(userId, msg);
}
logger.info(Thread.currentThread().getName() + "------------send to jms End");
}
@RequestMapping(value = "/ReceiveMessage", method = RequestMethod.GET)
@ResponseBody
public Object receive(@RequestParam(value = "userId",required=false)String userId) {
logger.info(Thread.currentThread().getName() + "------------receive from jms Start");
String tm=null;
if (userId==null||"".equals(userId)) {
tm = consumer.receive(destination);
} else {
tm = consumer.receive(userId);
}
logger.info(Thread.currentThread().getName() + "------------receive from jms End");
return tm.toString();
}
}
2.5 配置監(jiān)聽器(ek)
如果在配置文件中打開了監(jiān)聽器的注釋父阻，即打開監(jiān)聽器，消費者會立即去消費消息望抽，則還需要添加如下代碼：
public class QueueMessageListener implements MessageListener{
@Override
public void onMessage(Message message) {
TextMessage tm=(TextMessage) message;
try {
System.out.println("QueueMessageListener監(jiān)聽到了文本消息：\t"
+ tm.getText());
//do other work
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
3. 測試
啟動tomcat至非，將Javaweb項目運行在tomcat中，通過postman測試接口和方法
接受消息接口：http://localhost:8080/{project_neme}/mq/ReceiveMessage?userId={消息隊列名稱}
發(fā)送消息接口：http://localhost:8080/{project_neme}/mq/SendMessage?userId={消息隊列名稱}&msg={參數(shù)}
