浅谈jms之(通过spring整合activeMQ实现jms)实例
2015-01-26 16:51
886 查看
上篇说到使用activemq 实现jms消息服务 的实例,但是那个是没有spring进行整合和管理的;其实spring完整提供对jms的支持,所以我们可以通过spring来管理整合activemq 实现jms消息传递服务。
1.创建maven工程(我的项目都是通过maven管理的),在pom中引入我们要用的jar包,如activemq-core、spring-web、还有spring对jms支持的jar包等,如果不是通过maven 管理的 可以去spring官网下载相应的jar包。
2. ①按上篇中的开发流程,无论是对于一个queue队列模式还是 topic 模式的 , 首先需要连接activemq服务的connectionFactory。然后我们需要创建一个queue队列或一个topic 主题(destination 消息发送的目的站)。还有就是spring -jms 的api为我们提供了JmsTemplate,查看了 JmsTemplate类发现 spring已经将我们需要的 获取activemq 的连接
发送 都做了层封装;所以我们使用的时候只用调用模板类中的方法 就可以了 。则我们要在spring-bean.xml(applicationContext..xml)配置文件中配置上这些:
queue 模式:
②.接下来我们要创建queue队列的生产者 和消费者 :
QueueProducer.java
TopicPublisher.java
生产者测试类:
消费者测试类:
同样写个topic 的pub\sub测试类,这边要注意,我们之前说过topic模式和queue是有区别的,所以topic 要先运行订阅者测试类,然后再运行发布者测试类 才可以收到消息。
(3)之前我们讲的这些直接使用spring给的JMS template都是同步的收发消息 ,可是spring给我们提供另外一种的jms消息传递方式:message listener container,这个就是用于异步收发消息方式的。就是写一个实现MessageListener接口的类。这样消费者可以时时接受到生产者的消息,订阅者可以时时接受到发布者的消息.不用循环接受(异步接收消息)。
①首先我们要先实现消息监听接口的消费者实现类:
1.创建maven工程(我的项目都是通过maven管理的),在pom中引入我们要用的jar包,如activemq-core、spring-web、还有spring对jms支持的jar包等,如果不是通过maven 管理的 可以去spring官网下载相应的jar包。
<pre name="code" class="html"><dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-core</artifactId> <version>5.7.0</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>3.0.5.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-web</artifactId> <version>3.0.5.RELEASE</version> </dependency>
2. ①按上篇中的开发流程,无论是对于一个queue队列模式还是 topic 模式的 , 首先需要连接activemq服务的connectionFactory。然后我们需要创建一个queue队列或一个topic 主题(destination 消息发送的目的站)。还有就是spring -jms 的api为我们提供了JmsTemplate,查看了 JmsTemplate类发现 spring已经将我们需要的 获取activemq 的连接
发送 都做了层封装;所以我们使用的时候只用调用模板类中的方法 就可以了 。则我们要在spring-bean.xml(applicationContext..xml)配置文件中配置上这些:
queue 模式:
<!-- 配置jms 连接mq 的工厂类 --> <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616"/> </bean> <!-- (点到点的队列 )queueDest:接收消息的目的站 --> <bean id="queueDest" class="org.apache.activemq.command.ActiveMQQueue" > <constructor-arg index="0" value="spring-activemq-queue"/> </bean> <!-- 使用模板JMSTemplate : 在Spring框架中使用JMS传递消息有两种方式:JMS template 和message listener container, 前者用于同步收发消息,后者用于异步收发消息。 --> <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory"/> <property name="defaultDestination" ref="queueDest"/> </bean>topic 模式:
<!-- pub/sub的destination:接收订阅消息的目的站 --> <bean id="topicDest" class="org.apache.activemq.command.ActiveMQTopic"> <!-- 设置主题名称 --> <constructor-arg index="0" value="spring-activemq-topic"/> </bean> <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory" ref="connectionFactory" /> <property name="defaultDestination" ref="topicDest" /> <!-- 配置是否为发布订阅者模式,默认为false --> <property name="pubSubDomain" value="true"/> </bean>
②.接下来我们要创建queue队列的生产者 和消费者 :
QueueProducer.java
/** * 点对点 生产者 * @author leo * */ public class QueueProducer { /** * jms 模板 */ private JmsTemplate jmsTemplate; /** * queuedest 消息发送目的站 */ private Destination destination; public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public void setDestination(Destination destination) { this.destination = destination; } /** * 生产者发送消息 */ public void sendMsg(){ //消息构造器 MessageCreator creator = new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { TextMessage message =session.createTextMessage(); message.setText("发送点到点类型的信息"); return message; } }; jmsTemplate.send(this.destination, creator); } }QueueCustomer.java
/** * point -to -point 消费者 * * @author leo * */ public class QueueCustomer { //jms 模板 private JmsTemplate jmsTemplate; //目的地 private Destination destination; public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public void setDestination(Destination destination) { this.destination = destination; } /** * 接收消息的方法 */ public void receiveMsg(){ //接收来自于目的站的信息 TextMessage message = (TextMessage) jmsTemplate.receive(destination); try { System.out.println(message.getText()); } catch (JMSException e) { e.printStackTrace(); } } }以及applicationContext.xml 中配置管理生产者 和消费者:
<!-- 配置点到点的 消费者(Customer) 和生产者(Producer) --> <bean id="queueProducer" class="com.deppon.activemq.queue.producer.QueueProducer"> <property name="jmsTemplate" ref="jmsQueueTemplate"/> <property name="destination" ref="queueDest"/> <!--<property name="destination" ref="queueCustomerListenerDest"/> --> </bean> <bean id="queueCustomer" class="com.deppon.activemq.queue.customer.QueueCustomer"> <property name="jmsTemplate" ref="jmsQueueTemplate"/> <property name="destination" ref="queueDest"/> </bean>③.对于topic主题模式的 我们要创建他的发布者和订阅者:
TopicPublisher.java
/** * 发布/订阅 模式的 发布端 * @author leo * */ public class TopicPublisher { //jms模板 private JmsTemplate jmsTemplate; //目的站 private Destination destination; public void setDestination(Destination destination) { this.destination = destination; } public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } /** * 发送消息的方法 */ public void sendTopicMsg(){ //消息构造器 MessageCreator creator = new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { //创建消息 TextMessage message = session.createTextMessage(); message.setText("发送topic类型的消息-leo"); return message; } }; //发送消息 jmsTemplate.send(destination, creator); } }TopicSubscriber.java
/** * topic的 订阅端 * @author leo * */ public class TopicSubscriber { //jms模板 private JmsTemplate jmsTemplate; //目的站 private Destination destination; public void setDestination(Destination destination) { this.destination = destination; } public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } /** * 接收消息 */ public void receiveTopicMsg(){ TextMessage message =(TextMessage) jmsTemplate.receive(destination); try { System.out.println("接收topic消息:"+message.getText()); } catch (JMSException e) { e.printStackTrace(); } } }以及在applicationContext.xml 配置文件中关于订阅\发布的管理
<!-- 配置pub/sub的 发布者(subscriber) 和 订阅者(publisher) --> <bean id="topicPublisher" class="com.deppon.activemq.topic.publisher.TopicPublisher"> <property name="jmsTemplate" ref="jmsTopicTemplate"/> <property name="destination" ref="topicDest"/> <!-- <property name="destination" ref="topicListenerDest"/> --> </bean> <bean id="topicSubscriber" class="com.deppon.activemq.topic.subscriber.TopicSubscriber"> <property name="jmsTemplate" ref="jmsTopicTemplate"/> <property name="destination" ref="topicDest"/> </bean>那两个模式的关于spring 整合activemq 实现jms消息服务的简单demo 就完成了。接下来写两个测试测试下两种方式,看是消息传递是否成功。值得注意的是:activemq 服务一定要记得开启。
生产者测试类:
/** * 队列--生产者测试类 * @author leo * */ public class TestProducer { private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "/com/deppon/activemq/META-INF/ds-spring.xml"); /** * @param args */ public static void main(String[] args) { //获取生产者 QueueProducer producer =(QueueProducer) appContext.getBean("queueProducer"); //发送消息 producer.sendMsg(); } }
消费者测试类:
/** * 消费者测试类 * @author leo * */ public class TestCustomer { private static ApplicationContext appContext = new ClassPathXmlApplicationContext( "/com/deppon/activemq/META-INF/ds-spring.xml"); /** * @param args */ public static void main(String[] args) { //消费者 QueueCustomer customer = (QueueCustomer) appContext.getBean("queueCustomer"); //开启接收主线程 customer.receiveMsg(); } }运行生产者测试类,发现active上已经有新的队列,然后再运行消费者测试类,发现消息被取到。
同样写个topic 的pub\sub测试类,这边要注意,我们之前说过topic模式和queue是有区别的,所以topic 要先运行订阅者测试类,然后再运行发布者测试类 才可以收到消息。
(3)之前我们讲的这些直接使用spring给的JMS template都是同步的收发消息 ,可是spring给我们提供另外一种的jms消息传递方式:message listener container,这个就是用于异步收发消息方式的。就是写一个实现MessageListener接口的类。这样消费者可以时时接受到生产者的消息,订阅者可以时时接受到发布者的消息.不用循环接受(异步接收消息)。
①首先我们要先实现消息监听接口的消费者实现类:
/** * 消费者Customer实现text消息监听,实现(异步)时时接收 * @author leo * */ public class QueueCustomerListener implements MessageListener{ @Override public void onMessage(Message message) { TextMessage textMessage =null; try { //然后类型是属于textMessage if (message instanceof TextMessage) { textMessage =(TextMessage)message; System.out.println("队列 MessageListener收到信息:"+textMessage.getText()); } } catch (JMSException e) { e.printStackTrace(); } } }以及applicationContext.xml 配置文件中需要配置上相应的监听管理 以及要改掉生产者中所对应的队列目的站:
<bean id="queueCustomerListener" class="com.deppon.activemq.queue.linsteners.QueueCustomerListener"></bean> <!-- 新建一个队列, --> <bean id="queueCustomerListenerDest" class="org.apache.activemq.command.ActiveMQQueue" > <constructor-arg index="0" value="new-queue"/> </bean> <!-- 消息监听容器 --> <bean id="messageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="destination" ref="queueCustomerListenerDest"/> <property name="connectionFactory" ref="connectionFactory"/> <property name="messageListener" ref="queueCustomerListener"/> </bean>
这会不需要之前的那个消费者类:只要修改掉生产者中的目的站就可以
<bean id="queueProducer" class="com.deppon.activemq.queue.producer.QueueProducer"> <property name="jmsTemplate" ref="jmsQueueTemplate"/> <!-- <property name="destination" ref="queueDest"/> --> <property name="destination" ref="queueCustomerListenerDest"/> </bean>②.对于topic主题模式的,也是一样:
/** * 订阅者Subscriber实现text消息监听,实现(异步)时时接收 * @author leo * */ public class TopicSubscriberListener implements MessageListener{ @Override public void onMessage(Message message) { TextMessage textMessage =(TextMessage)message; try { System.out.println("接收实现MessageListener的topic监听:"+textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } }以及applicationContext.xml 配置文件中需要配置上相应的监听管理 以及要改掉生发布者中所对应的主题目的站:
<!-- 订阅监听 --> <bean id="topicSubscriberListener" class="com.deppon.activemq.topic.listeners.TopicSubscriberListener"></bean> <!-- 新建一个主题(平台) --> <bean id="topicListenerDest" class="org.apache.activemq.command.ActiveMQTopic"> <!-- 设置队列名称 --> <constructor-arg index="0" value="new-topic"/> </bean> <bean id="myTopicListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="destination" ref="topicListenerDest"/> <property name="connectionFactory" ref="connectionFactory"/> <property name="messageListener" ref="topicSubscriberListener"/> <!-- 配置是否为发布订阅者模式,默认为false --> <property name="pubSubDomain" value="true"/> </bean>发布者中原来的目的站,要改成新的主题目的站:
<bean id="topicPublisher" class="com.deppon.activemq.topic.publisher.TopicPublisher"> <property name="jmsTemplate" ref="jmsTopicTemplate"/> <!-- <property name="destination" ref="topicDest"/> --> <property name="destination" ref="topicListenerDest"/> </bean>③然后启动一个之前写的那个生产者测试类,发现运行生产者测试类后 不用再运行消费者测试类就可以收到消息。同样topic 也是如此。
相关文章推荐
- 通过spring整合activeMQ实现jms实例
- 浅谈jms之(用activemq实现jms实例)
- Spring整合JMS-基于activeMQ实现(一)
- Spring整合JMS(一)——基于ActiveMQ实现
- spring整合JMS - 基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- Spring整合JMS-基于activeMQ实现(二)
- 深入浅出JMS(四)--Spring和ActiveMQ整合的完整实例
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合Jms学习(一)_基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- Spring整合JMS——基于ActiveMQ实现
- 深入浅出JMS(四)--Spring和ActiveMQ整合的完整实例
- Spring整合JMS——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现
- Spring整合JMS(一)——基于ActiveMQ实现