ActiveMQ和spring集成

1.消息发送类

Java代码







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;

}

}

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.消息接收类，此处采用监听器方式实现

Java代码







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.发送返回消息(可省略)
}
}

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.消息发送接口

Java代码







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;

}

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配置文件

Java代码







<?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>

<?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>