ActiveMQ实战(五)--Spring整合ActiveMQ实现点对点与主题发布订阅通信
2017-07-23 20:58
429 查看
一、项目结构
二、加入Maven依赖
三、配置文件
applicationContext.xml
spring-mvc.xml
ActiveMQ.xml
web.xml
四、一些类
两个Controller类,一个用来测试点对点通信,另一个是用来测试点对点与主题发布订阅通信:
点对点通信控制类:
主题发布订阅通信:
一个消费者里面包含queue和topic
queue包下有QueueReceiver1和QueueReceiver2
QueueReceiver1
QueueReceiver2
topic包下有TopicReceiver1,TopicReceiver2
TopicReceiver1
TopicReceiver2
五、测试
A.点对点通信
1)手动测试点对点通信
首先运行项目,然后访问: http://localhost:8080/Spring_ActiveMQ/sendMessage.jsp
输入要发送的内容,点击发送,控制台打印如下
ActiveMQ控制台会多出一条记录
然后访问地址:http://localhost:8080/Spring_ActiveMQ/ReceiveMessage.jsp 接受消息
点击"接受消息",控制台会打印出生产者发送的消息
2)自动监听点对点通信
在实际项目中,我们很少会自己手动去获取消息,而是使用一个监听器来监听是否有消息到达,这样子可以很快的完成对消息的处理。
在ActiveMQ.xml配置文件中有段代码是用来配置监听器的
QueueMessageListener类为
这样的话,我们在发消息的时候,后台就会自动的监听
注意:当开启自动监听以后,手动访问就会失效。
B.主题发布订阅通信
当访问: http://localhost:8080/Spring_ActiveMQ/index.jsp
输入要发送的内容,然后点击发送Queue消息,会发现后台只有一个接受到了信息
然后发送Topic消息,后台两个都收到了信息
从上面可看出队列模型和PUB/SUB模型的区别,Queue只能由一个消费者接收,其他Queue中的成员无法接受到被已消费的信息,而Topic则可以,只要是订阅了Topic的消费者,全部可以获取到生产者发布的信息。
完整示例: http://download.csdn.net/detail/u011781521/9908153
二、加入Maven依赖
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.fendo</groupId> <artifactId>Spring_ActiveMQ</artifactId> <packaging>war</packaging> <version>0.0.1-SNAPSHOT</version> <name>Spring_ActiveMQ Maven Webapp</name> <url>http://maven.apache.org</url> <!-- 版本管理 --> <properties> <springframework>4.1.8.RELEASE</springframework> <javax.servlet>3.1.0</javax.servlet> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.10</version> <scope>test</scope> </dependency> <dependency> <groupId>jstl</groupId> <artifactId>jstl</artifactId> <version>1.2</version> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <version>${javax.servlet}</version> </dependency> <dependency> <groupId>log4j</groupId> <artifactId>log4j</artifactId> <version>1.2.16</version> </dependency> <!-- spring --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-core</artifactId> <version>${springframework}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>${springframework}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-tx</artifactId> <version>${springframework}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-webmvc</artifactId> <version>${springframework}</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>${springframework}</version> </dependency> <!-- xbean 如<amq:connectionFactory /> --> <dependency> <groupId>org.apache.xbean</groupId> <artifactId>xbean-spring</artifactId> <version>3.16</version> </dependency> <!-- activemq --> <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> </dependencies> <build> <finalName>Spring_ActiveMQ</finalName> </build> </project>
三、配置文件
applicationContext.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:jms="http://www.springframework.org/schema/jms" 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-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.14.3.xsd"> <bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor"/> <!-- 配置扫描路径 --> <context:component-scan base-package="com.fendo"> <!-- 只扫描Service,也可以添加Repostory,但是要把Controller排除在外,Controller由spring-mvc.xml去加载 --> <context:exclude-filter type="annotation" expression="org.springframework.stereotype.Controller"/> </context:component-scan> </beans>
spring-mvc.xml
<?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" > <!-- 这里只扫描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="/" /> <property name="suffix" value=".jsp" /> <!-- 定义其解析视图的order顺序为1 --> <property name="order" value="1" /> </bean> </beans>
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:jms="http://www.springframework.org/schema/jms" 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-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.1.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd"> <!-- ActiveMQ 连接工厂 --> <!-- 如果连接网络:tcp://ip:61616;未连接网络:tcp://localhost:61616 以及用户名,密码--> <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin" /> <!-- 配置JMS连接工厂 --> <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory --> <constructor-arg ref="amqConnectionFactory" /> <!-- Session缓存数量 --> <property name="sessionCacheSize" value="100" /> </bean> <!-- 定义消息队列(Queue) --> <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <!-- 设置消息队列的名字 --> <constructor-arg> <value>fendo</value> </constructor-arg> </bean> <!-- Spring JmsTemplate 的消息生产者 start--> <!-- 定义JmsTemplate的Queue类型 --> <!-- 配置JMS模板(Queue),Spring提供的JMS工具类,它发送、接收消息。 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory" /> <property name="defaultDestination" ref="demoQueueDestination" /> <property name="receiveTimeout" value="10000" /> <!-- 非pub/sub模型(发布/订阅),即队列模式 --> <!-- 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" /> <!-- pub/sub模型(发布/订阅) --> <property name="pubSubDomain" value="true" /> </bean> <!--Spring JmsTemplate 的消息生产者 end--> <!-- 消息消费者 start--> <!-- 定义Queue监听器 --> <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <jms:listener destination="test.queue" ref="queueReceiver1"/> <jms:listener destination="test.queue" ref="queueReceiver2"/> </jms:listener-container> <!-- 定义Topic监听器 --> <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto"> <jms:listener destination="test.topic" ref="topicReceiver1"/> <jms:listener destination="test.topic" ref="topicReceiver2"/> </jms:listener-container> <!-- 消息消费者 end --> <!-- 配置消息队列监听者(Queue) --> <!-- <bean id="queueMessageListener" class="com.fendo.filter.QueueMessageListener" /> --> <!-- 显示注入消息监听容器(Queue),配置连接工厂,监听的目标是demoQueueDestination,监听器是上面定义的监听器--> <!-- <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>
web.xml
<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd" version="3.0"> <display-name>Archetype Created Web Application</display-name> <!-- Log4J Start --> <context-param> <param-name>log4jConfigLocation</param-name> <param-value>classpath:log4j.properties</param-value> </context-param> <context-param> <param-name>log4jRefreshInterval</param-name> <param-value>6000</param-value> </context-param> <!-- Spring Log4J config --> <listener> <listener-class>org.springframework.web.util.Log4jConfigListener</listener-class> </listener> <!-- Log4J End --> <!-- 加载spring的配置文件,例如hibernate、jms等集成 --> <context-param> <param-name>contextConfigLocation</param-name> <param-value> classpath*:applicationContext.xml,classpath*:ActiveMQ.xml </param-value> </context-param> <listener> <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class> </listener> <servlet> <servlet-name>springMVC</servlet-name> <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class> <init-param> <param-name>contextConfigLocation</param-name> <param-value>classpath:spring-mvc.xml</param-value> </init-param> <load-on-startup>1</load-on-startup> </servlet> <servlet-mapping> <servlet-name>springMVC</servlet-name> <url-pattern>/</url-pattern> </servlet-mapping> <!-- 处理编码格式 --> <filter> <filter-name>characterEncodingFilter</filter-name> <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class> <init-param> <param-name>encoding</param-name> <param-value>UTF-8</param-value> </init-param> <init-param> <param-name>forceEncoding</param-name> <param-value>true</param-value> </init-param> </filter> <filter-mapping> <filter-name>characterEncodingFilter</filter-name> <url-pattern>/*</url-pattern> </filter-mapping> </web-app>
四、一些类
两个Controller类,一个用来测试点对点通信,另一个是用来测试点对点与主题发布订阅通信:
点对点通信控制类:
package com.fendo.controller; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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.ResponseBody; import com.fendo.service.ConsumerService; import com.fendo.service.ProducerService; import javax.annotation.Resource; import javax.jms.Destination; import javax.jms.TextMessage; /** * * @author fendo * */ @Controller public class MessageController { private Logger logger = LoggerFactory.getLogger(MessageController.class); @Resource(name = "demoQueueDestination") private Destination destination; //队列消息生产者 @Resource(name = "producerService") private ProducerService producer; //队列消息消费者 @Resource(name = "consumerService") private ConsumerService consumer; @RequestMapping(value = "/SendMessage", method = RequestMethod.POST) @ResponseBody public void send(String msg) { logger.info(Thread.currentThread().getName()+"------------send to jms Start"); producer.sendMessage(msg); logger.info(Thread.currentThread().getName()+"------------send to jms End"); } @RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET) @ResponseBody public Object receive(){ logger.info(Thread.currentThread().getName()+"------------receive from jms Start"); TextMessage tm = consumer.receive(destination); logger.info(Thread.currentThread().getName()+"------------receive from jms End"); return tm; } }
主题发布订阅通信:
package com.fendo.controller; import javax.annotation.Resource; 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.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import com.fendo.producer.queue.QueueSender; import com.fendo.producer.topic.TopicSender; /** * * @author fendo * @description controller测试 */ @Controller @RequestMapping("/activemq") public class ActivemqController { @Autowired QueueSender queueSender; @Autowired TopicSender topicSender; /** * 发送消息到队列 * Queue队列:仅有一个订阅者会收到消息,消息一旦被处理就不会存在队列中 * @param message * @return String */ @ResponseBody @RequestMapping("queueSender") public String queueSender(@RequestParam("message")String message){ String opt=""; try { queueSender.send("test.queue", message); opt = "suc"; } catch (Exception e) { opt = e.getCause().toString(); } return opt; } /** * 发送消息到主题 * Topic主题 :放入一个消息,所有订阅者都会收到 * 这个是主题目的地是一对多的 * @param message * @return String */ @ResponseBody @RequestMapping("topicSender") public String topicSender(@RequestParam("message")String message){ String opt = ""; try { topicSender.send("test.topic", message); opt = "suc"; } catch (Exception e) { opt = e.getCause().toString(); } return opt; } }
一个消费者里面包含queue和topic
queue包下有QueueReceiver1和QueueReceiver2
QueueReceiver1
package com.fendo.consumer.queue; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; import org.springframework.stereotype.Component; /** * * @author fendo * @description 队列消息监听器 * */ @Component public class QueueReceiver1 implements MessageListener { @Override public void onMessage(Message message) { try { System.out.println("QueueReceiver1接收到消息:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
QueueReceiver2
package com.fendo.consumer.queue; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; import org.springframework.stereotype.Component; /** * * @author fendo * @description 队列消息监听器 * */ @Component public class QueueReceiver2 implements MessageListener { @Override public void onMessage(Message message) { try { System.out.println("QueueReceiver2接收到消息:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
topic包下有TopicReceiver1,TopicReceiver2
TopicReceiver1
package com.fendo.consumer.topic; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; import org.springframework.stereotype.Component; /** * * @author fendo * @description Topic消息监听器 * */ @Component public class TopicReceiver1 implements MessageListener{ @Override public void onMessage(Message message) { try { System.out.println("TopicReceiver1接收到消息:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
TopicReceiver2
package com.fendo.consumer.topic; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; import org.springframework.stereotype.Component; /** * * @author fendo * @description Topic消息监听器 * */ @Component public class TopicReceiver2 implements MessageListener{ @Override public void onMessage(Message message) { try { System.out.println("TopicReceiver2接收到消息:"+((TextMessage)message).getText()); } catch (JMSException e) { e.printStackTrace(); } } }
五、测试
A.点对点通信
1)手动测试点对点通信
首先运行项目,然后访问: http://localhost:8080/Spring_ActiveMQ/sendMessage.jsp
输入要发送的内容,点击发送,控制台打印如下
ActiveMQ控制台会多出一条记录
然后访问地址:http://localhost:8080/Spring_ActiveMQ/ReceiveMessage.jsp 接受消息
点击"接受消息",控制台会打印出生产者发送的消息
2)自动监听点对点通信
在实际项目中,我们很少会自己手动去获取消息,而是使用一个监听器来监听是否有消息到达,这样子可以很快的完成对消息的处理。
在ActiveMQ.xml配置文件中有段代码是用来配置监听器的
<!-- 配置消息队列监听者(Queue) --> <bean id="queueMessageListener" class="com.fendo.filter.QueueMessageListener" /> <!-- 显示注入消息监听容器(Queue),配置连接工厂,监听的目标是demoQueueDestination,监听器是上面定义的监听器 --> <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>
QueueMessageListener类为
package com.fendo.filter; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage; /** * * @author fendo * */ public class QueueMessageListener implements MessageListener { public void onMessage(Message message) { TextMessage tm = (TextMessage) message; try { System.out.println("QueueMessageListener监听到了文本消息:\t" + tm.getText()); } catch (JMSException e) { e.printStackTrace(); } } }
这样的话,我们在发消息的时候,后台就会自动的监听
注意:当开启自动监听以后,手动访问就会失效。
B.主题发布订阅通信
当访问: http://localhost:8080/Spring_ActiveMQ/index.jsp
输入要发送的内容,然后点击发送Queue消息,会发现后台只有一个接受到了信息
然后发送Topic消息,后台两个都收到了信息
从上面可看出队列模型和PUB/SUB模型的区别,Queue只能由一个消费者接收,其他Queue中的成员无法接受到被已消费的信息,而Topic则可以,只要是订阅了Topic的消费者,全部可以获取到生产者发布的信息。
完整示例: http://download.csdn.net/detail/u011781521/9908153
相关文章推荐
- ActiveMQ实战(六)--Spring整合ActiveMQ实现队列和主题发布订阅通信
- activeMQ 点对点以及发布与订阅 - 以及spring的整合&集群方式
- 实战day09(一)----Activemq整合spring
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- spring整合JMS - 基于ActiveMQ实现
- 消息中间件-activemq实战之整合Spring(四)
- Spring整合JMS——基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- 分布式缓存技术redis学习系列(五)——redis实战(redis与spring整合,分布式锁实现)
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现 (转)
- 消息中间件-activemq实战之整合Spring(四)
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS-基于activeMQ实现(一)
- Spring整合JMS(一)——基于ActiveMQ实现
- spring整合JMS实现同步收发消息(基于ActiveMQ的实现)