1.消息发送类
package cn.com.biceng.jms.queue; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import org.apache.activemq.command.ActiveMQQueue; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import cn.com.biceng.jms.IQueueSender; /** * 队列消息发送者。 */ public class QueueSender implements IQueueSender { // 发送消息的目的地 private Destination destination; private JmsTemplate jmsTemplate; public QueueSender(ActiveMQQueue queue){ this.destination = queue; } /** * 发送消息。 * * @param message * 报文。 * @return * @author ma_chunlin * @date 2013-5-6 */ public void send(final String message) throws JMSException { jmsTemplate.send(destination, new MessageCreator() { public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } }
2.消息接收类,此处采用监听器方式实现
package cn.com.biceng.jms.queue; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.log4j.Logger; import org.springframework.jms.listener.SessionAwareMessageListener; /** * * 消息接收监听器。 */ public class QueueReceiverListener implements SessionAwareMessageListener { private static Logger log = Logger.getLogger(QueueReceiverListener.class); public void onMessage(Message message, Session session) throws JMSException { //1.接收报文 TextMessage msg = (TextMessage)message; if(log.isDebugEnabled()){ log.debug("QueueReceiverListener接收到报文:" + msg.getText()); } //2.解析报文,建议此处使用工具类解析 //3.调用接口,处理业务,建议此处使用“抽象工厂模式”调用内部接口,以使代码符合“开-闭原则” //4.发送返回消息(可省略) } }
3.消息发送接口
package cn.com.biceng.jms; import javax.jms.JMSException; /** * 队列消息发送者。 * */ public interface IQueueSender { /** * 发送消息。 * * @param message * 报文。 * @return * @author ma_chunlin * @date 2013-5-6 */ public abstract void send(final String message) throws JMSException; }
4. 用到的jar
5. spring配置文件
<?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:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd" default-autowire="byName"> <!-- JMS activeMq的配置 --> <!-- 1. 配置connectionFactory --> <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory"> <bean class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL"> <!-- JMS消息服务器的IP和端口号 --> <value>tcp://192.168.8.241:61616</value> </property> </bean> </property> <property name="maxConnections" value="100"/> <!-- <property name="idleTimeout" value="60"/> --> </bean> <!-- 2.Spring JMS Template --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <property name="connectionFactory"> <ref local="jmsFactory" /> </property> <property name="defaultDestinationName" value="queue_eSign_call_feeSys" /> <!-- 区别它采用的模式为false是p2p, true是订阅 --> <property name="pubSubDomain" value="false" /> </bean> <!-- 3.发送给计费系统的消息 --> <bean id="queue_eSign_to_feeSys" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg index="0" value="queue_eSign_to_feeSys" /> </bean> <!-- 4.来自计费系统的消息 --> <bean id="queue_feeSys_to_eSign" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg index="0" value="queue_feeSys_to_eSign" /> </bean> <!-- 5.消息发送者 --> <bean id="QueueSender" class="cn.com.biceng.jms.queue.QueueSender"> <constructor-arg ref="queue_eSign_to_feeSys" /> <property name="jmsTemplate" ref="jmsTemplate" /> </bean> <!-- 6.队列监听器 --> <bean id="QueueReceiverListener" class="cn.com.biceng.jms.queue.QueueReceiverListener"> </bean> <!-- 7.队列监听器的容器 --> <bean id="QueueReceiverContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="concurrentConsumers" value="1" /> <property name="connectionFactory" ref="jmsFactory" /> <!-- 监听的队列 --> <property name="destinationName" value="queue_feeSys_to_eSign" /> <property name="messageListener" ref="QueueReceiverListener" /> <property name="pubSubNoLocal" value="false"/> </bean> </beans>