实现一个基于SpringMVC+JMS+ActiveMQ+Tomcat+JDK1.8+IDEA工具 ,Spring4.1.0和ActiveMQ5.15整合的实例,实现PTP和订阅/发布两种消息模型
一 添加依赖
这里我贴上我所有的依赖,其中有个坑就是springaop要用高版本,否则会报方法找不到,主要就是spring提供的jms依赖
<dependencies> <!--activeMQ--> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-all</artifactId> <version>5.15.5</version> </dependency> <!--spring--> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>${org.springframework-version}</version> </dependency> <!--大坑,aop千万要用高版本的,否则会一直报错--> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-aop</artifactId> <version>4.3.12.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-beans</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jdbc</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-orm</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-test</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.aspectj</groupId> <artifactId>aspectjweaver</artifactId> <version>1.8.1</version> </dependency> <dependency> <groupId>org.aspectj</groupId> <artifactId>aspectjrt</artifactId> <version>1.8.1</version> </dependency> <!-- springmvc --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-tx</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-expression</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> <version>${org.springframework-version}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> <version>${org.springframework-version}</version> </dependency> <!-- logging --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>${org.slf4j-version}</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>jcl-over-slf4j</artifactId> <version>${org.slf4j-version}</version> </dependency> <!-- log4j --> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.17</version> </dependency> <!--spring-JMS Spring提供了对JMS的支持,需要添加Spring支持jms的包 --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>4.1.0.RELEASE</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.11</version> <scope>test</scope> </dependency> </dependencies>
二 设计目录结构
三 整合文件配置(重点)
主要是mq整合配置,需要注意的就是头文件要加上引用的,这里都贴上:
spring配置:
<?xml version="1.0" encoding="UTF-8"?> <!-- 查找最新的schemaLocation 访问 http://www.springframework.org/schema/ --> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.9.0.xsd"> <!-- 开启注解 --> <context:annotation-config></context:annotation-config> <bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor"/> <!-- 配置扫描路径 --> <context:component-scan base-package="com.hou"> <!--扫描serveice等注解的类--> <context:include-filter type="annotation" expression="org.springframework.stereotype.Service" /> <context:include-filter type="annotation" expression="org.springframework.stereotype.Repository" /> <context:include-filter type="annotation" expression="org.springframework.stereotype.Component" /> <!--排除Controller,Controller由springmvc加载--> <context:exclude-filter type="annotation" expression="org.springframework.stereotype.Controller"/> </context:component-scan> </beans>
springmvc配置:
<?xml version="1.0" encoding="UTF-8"?> <!-- 查找最新的schemaLocation 访问 http://www.springframework.org/schema/ --> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.0.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.0.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.0.xsd"> <!-- 启用MVC注解 --> <mvc:annotation-driven /> <!-- 静态资源文件,不会被Spring MVC拦截 --> <mvc:resources location="/resources/" mapping="/resources/**"/> <!-- 指定Sping组件扫描的基本包路径 --> <context:component-scan base-package="com.hou" > <!-- 这里只扫描Controller,不可重复加载Service --> <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/> </context:component-scan> <!-- JSP视图解析器--> <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver"> <property name="prefix" value="/WEB-INF/" /> <property name="suffix" value=".jsp" /> <!-- 定义其解析视图的order顺序为1 --> <property name="order" value="1" /> </bean> </beans>
mq整合配置:
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd"> <!-- ActiveMQ 连接工厂 --> <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://localhost:61616" userName="houzheng" password="houzheng" /> <!-- Spring Caching连接工厂 --> <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!--JMS 连接工厂--> <constructor-arg ref="amqConnectionFactory" /> <!-- Session缓存数量 --> <property name="sessionCacheSize" value="100" /> </bean> <!--生产者配置Bean,JMS模板(Queue),Spring提供的JMS工具类,它发送、接收消息--> <!-- 定义JmsTemplate的Queue类型 --> <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <constructor-arg ref="connectionFactory" /> <!--可配置默认的Destination--> <!--<property name="defaultDestination" ref="demoQueueDestination" />--> <!--接受超时--> <!--<property name="receiveTimeout" value="10000" />--> <!-- true是topic,false是queue,默认是false,此处显示写出false --> <property name="pubSubDomain" value="false" /> </bean> <!-- 定义JmsTemplate的Topic类型 --> <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <constructor-arg ref="connectionFactory" /> <property name="pubSubDomain" value="true" /> </bean> <!-- 定义Queue监听器 --> <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <!--配置每一个消费者的消息监听处理--> <jms:listener destination="queue" ref="queueReceiver01"/> <jms:listener destination="queue" ref="queueReceiver02"/> </jms:listener-container> <!-- 定义Topic监听器 --> <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <jms:listener destination="topic" ref="topicReceiver01"/> <jms:listener destination="topic" ref="topicReceiver02"/> </jms:listener-container> </beans>
四 开发java代码
首先开发消息发送类,因为通过xml与注解配置了bean,所以可直接注入jms模板发送消息:
@Service public class QueueSender { //当需要注入的实现类有多个时,可以通过@Qualifier指定注入我们需要的实现类,按名字 @Autowired @Qualifier("jmsQueueTemplate") private JmsTemplate jmsTemplate; /** * 发送一条消息到指定的队列(目标) * @param queueName 队列名称 * @param message 消息内容 */ public void send(String queueName,final String message){ jmsTemplate.send(queueName, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } }
主题发送类:
@Service public class TopicSender { @Autowired @Qualifier("jmsTopicTemplate") private JmsTemplate jmsTemplate; /** * 发送一条消息到指定的队列(目标) * @param queueName 队列名称 * @param message 消息内容 */ public void send(String topicName,final String message){ jmsTemplate.send(topicName, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } }
另外这里的jmsTemplate还可以根据需求设置是否持久化,优先级,超时,目的地等等
然后是消费者,因为配置了监听器,所以只需要实现监听接口就可以:
@Component public class QueueReceiver01 implements MessageListener { @Override public void onMessage(Message message) { System.out.println("1号消费者接受到队列的消息:"+message.toString()); } }
//实现MessageListener消息监听器,实际中肯定不可能主动去接受消息,所以一般是用监听器 @Component public class TopicReceiver01 implements MessageListener { @Override public void onMessage(Message message) { System.out.println("1号消费者接受到订阅的消息:"+message.toString()); } }
最后是controller:
package com.hou.springmq.controller; import com.hou.springmq.mq.producer.queue.QueueSender; import com.hou.springmq.mq.producer.topic.TopicSender; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; @Controller @RequestMapping("/mq") public class ActiveMQController { //注入发送类 @Autowired private QueueSender queueSender; @Autowired private TopicSender topicSender; //测试方法 @RequestMapping(value = "/test", method = RequestMethod.GET) @ResponseBody public String test01(){ System.out.println("----------"); return "houzheng"; } //发送队列 @ResponseBody @RequestMapping("/sendQueue") public String sendQueue(@RequestParam("message")String message){ //消息发送到指定的queue queueSender.send("queue",message); return "suc"; } //发送主题 @ResponseBody @RequestMapping("/sendTopic") public String sendTopic(@RequestParam("message")String message){ topicSender.send("topic",message); return "suc"; } }
五 启动mq,测试
topic所有消费者都会收到,而queue只有一个消费者会接受到!
最后,我这个整合主要是参考这个博客,部分代码也是直接拿过来学习用的,写的挺好:https://blog.csdn.net/jiuqiyuliang/article/details/48758203
因为一直用vue,html5 ,jsp都快忘光了,还有spring的一些配置,正好复习下,学习回顾!
源码:
最后放上源码下载地址,在我的码云里面,欢迎大家指正交流!
https://gitee.com/houzheng1216/architect/tree/master/
在里面的activemqspring中