spring整合ActiveMq：
1：依赖的jar包:
2:spring-activemq.xml 的配置：
代码：
<?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:amq="http://activemq.apache.org/schema/core"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:mvc="http://www.springframework.org/schema/mvc"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/mvc
http://www.springframework.org/schema/mvc/spring-mvc.xsd">
<context:component-scan base-package="com.floor.shop.service.mq" />
<mvc:annotation-driven />
<!--工厂对象-->
<amq:connectionFactory id="amqConnectionFactory"
brokerURL="tcp://192.168.1.20: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="activeMQQueue" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 设置消息队列的名字 -->
<constructor-arg name="name" value="FirstQueue"/>
</bean> <!-- 配置JMS模板（Queue），Spring提供的JMS工具类，它发送、接收消息。 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="activeMQQueue" />
<property name="receiveTimeout" value="10000" />
<!-- true是topic，false是queue，默认是false，此处显示写出false -->
<property name="pubSubDomain" value="false" />
</bean> </beans>
3：mq发送消息和接受消息的API封装：
3-1SendMsg:
3-1代码：
package com.floor.shop.service.mq; import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
@Service
public class SenderService {
/**
* Created by ause on 2017-09-26.
*/
@Autowired
private JmsTemplate jmsTemplate;
@Autowired
private ActiveMQQueue activeMQQueue;
/**
* 向配置文件中的默认队列发送消息
*/
public void sendMessage(final String msg) {
System.out.println("向队列" + activeMQQueue.toString() + "发送了消息------------" + msg);
jmsTemplate.send(activeMQQueue, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
/**
* 向指定队列发送消息
*/
public void sendMessage(final String msg,String queueName) {
ActiveMQQueue activeMQQueue = new ActiveMQQueue(queueName);
jmsTemplate.setDefaultDestination(activeMQQueue);
System.out.println("向队列" + activeMQQueue.toString() + "发送了消息------------" + msg);
jmsTemplate.send(activeMQQueue, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
}
3-2:receiveMsg:
3-2:代码：
package com.floor.shop.service.mq;
/**
* Created by ause on 2017-09-26.
*/ import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service; import javax.jms.JMSException;
import javax.jms.TextMessage; @Service
public class ReceiveService { @Autowired
private JmsTemplate jmsTemplate;
@Autowired
private ActiveMQQueue activeMQQueue;
/**
* 接收消息
*/
public TextMessage receive() {
TextMessage tm = (TextMessage) jmsTemplate.receive(activeMQQueue);
try {
if(tm!=null){
System.out.println("从队列" + activeMQQueue.toString() + "收到了消息：\t"
+ tm.getText());
}else {
System.out.println("队列为空！");
}
} catch (JMSException e) {
e.printStackTrace();
}
return tm;
}
/**
* 接收消息
*/
public TextMessage receive(String queueName) {
ActiveMQQueue activeMQQueue = new ActiveMQQueue(queueName);
TextMessage tm = (TextMessage) jmsTemplate.receive(activeMQQueue);
try {
if(tm!=null){
System.out.println("从队列" + activeMQQueue.toString() + "收到了消息：\t"
+ tm.getText());
}else {
System.out.println("队列为空！！");
} } catch (JMSException e) {
e.printStackTrace();
}
return tm;
} }
4：测试：
代码：
package com.floor.shop.test; import com.floor.shop.service.mq.ReceiveService;
import com.floor.shop.service.mq.SenderService;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import javax.jms.JMSException;
import javax.jms.TextMessage; @RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration("classpath:spring/spring-activeMq.xml")
public class TestMq {
@Autowired
private ReceiveService receiveService;
@Autowired
private SenderService senderService;
@Test
public void testSS(){
senderService.sendMessage("mq成功了");
}
@Test
public void testRs(){
TextMessage receive = receiveService.receive();
}
}
5：编写监听类：
在实际项目中，我们很少会自己手动去获取消息，如果需要手动去获取消息，那就没有必要使用到ActiveMq了，可以用一个Redis 就足够了。
不能手动去获取消息，那么我们就可以选择使用一个监听器来监听是否有消息到达，这样子可以很快的完成对消息的处理。相当于开启了事物一样，会
自动获取队列里面的消息。
首先我们需要创建一个类实现 MessageListener 接口：
代码：
package activeMq; import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject; import javax.jms.*; public class QueueMessageListener implements MessageListener {
@Override
public void onMessage(Message m) {
System.out.println(m.toString());
try {
if (m instanceof TextMessage) { //接收文本消息
TextMessage message = (TextMessage) m;
System.out.println("文本消息:" + message.getText());
} else if (m instanceof MapMessage) { //接收键值对消息
MapMessage message = (MapMessage) m;
// System.out.println(message.getLong("age"));
// System.out.println(message.getDouble("sarray"));
// System.out.println(message.getString("username"));
System.out.println("键值对消息 ");
} else if (m instanceof StreamMessage) { //接收流消息
StreamMessage message = (StreamMessage) m;
System.out.println("流消息:" + message.readString());
System.out.println("流消息:" + message.readLong());
} else if (m instanceof BytesMessage) { //接收字节消息
byte[] b = new byte[1024];
int len = -1;
BytesMessage message = (BytesMessage) m;
while ((len = message.readBytes(b)) != -1) {
System.out.println("字节消息:" + new String(b, 0, len));
String jsonstr = new String(b, 0, len);
JSONObject jsonObject = JSON.parseObject(jsonstr);
System.out.println("接收成功");
}
} else if (m instanceof ObjectMessage) { //接收对象消息
ObjectMessage message = (ObjectMessage) m;
// User user = (User)message.getObject();
System.out.println("对象消息");
} else {
System.out.println(m);
}
} catch (JMSException e) {
e.printStackTrace();
} }
}
然后在 spring-activemq.xml 中配置：
<!-- 消息侦听器 -->
<bean id="myMessageListener" class="activeMq.QueueMessageListener"></bean> <!-- 消息监听 -->
<bean id="listenerContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<!-- 消息监听器输出消息的数量 -->
<property name="connectionFactory" ref="connectionFactory" />
<property name="destinationName" value="firstQueue" />
<property name="messageListener" ref="myMessageListener" />
</bean>
就这么简单咯，快去试试吧！！！！！
也可以参考博客：https://www.cnblogs.com/jaycekon/p/ActiveMq.html
