在调用JMS消息消费者的receive()方法接收消息时，调用线程在消息可用之前一直阻塞。该线程出了等待还是等待，无所事事。这样的消息接收是同步消息接收，因为只用等到消息到达才能接收线程的工作。
有同步的消息接收就有异步的消息接收，异步的消息接收就是注册一个消息监听器，该消息监听器必须实现javax.jms.MessageListener接口，当消息到达时将调用onMessage()方法，以消息作为方法的参数。
原生的接口是javax.jms.MessageListener，除了这个原生的接口外，Spring 还提供了 SessionAwareMessageListener和MessageListenerAdapter.
/**
*
* @author zhangwei_david
* @version $Id: MessageListener.java, v 0.1 2015年1月31日 下午9:06:02 zhangwei_david Exp $
*/
public class MailMessageListener implements MessageListener {
private MessageConverter messageConverter;
/**
* @see javax.jms.MessageListener#onMessage(javax.jms.Message)
*/
public void onMessage(Message msg) {
try {
System.out.println("on message:" + messageConverter.fromMessage(msg));
} catch (JMSException e) {
}
}
/**
* Setter method for property <tt>messageConverter</tt>.
*
* @param messageConverter value to be assigned to property messageConverter
*/
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
}
}
Spring 中提供了多种消息监听器的容器，常用的容器有SimpleMessageListenerContainer 和DefaultMessageListenerContainer。SimpleMessageListenerContainer 是一个最简单的容器，不提供事务的支持，DefaultMessageListenerContainer是默认的容器实现，支持事务。
/**
*
* @author zhangwei_david
* @version $Id: ToStringBase.java, v 0.1 2015年2月2日 下午7:41:52 zhangwei_david Exp $
*/
public class ToStringBase {
/**
* @see java.lang.Object#toString()
*/
@Override
public String toString() {
return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
}
}
定义一个基本的对象
/**
*
* @author zhangwei_david
* @version $Id: Mail.java, v 0.1 2015年2月2日 下午7:25:24 zhangwei_david Exp $
*/
public class Mail extends ToStringBase {
/**id**/
private String mailId;
private String from;
private String to;
private String content;
/**
* Getter method for property <tt>mailId</tt>.
*
* @return property value of mailId
*/
public String getMailId() {
return mailId;
}
/**
* Setter method for property <tt>mailId</tt>.
*
* @param mailId value to be assigned to property mailId
*/
public void setMailId(String mailId) {
this.mailId = mailId;
}
/**
* Getter method for property <tt>from</tt>.
*
* @return property value of from
*/
public String getFrom() {
return from;
}
/**
* Setter method for property <tt>from</tt>.
*
* @param from value to be assigned to property from
*/
public void setFrom(String from) {
this.from = from;
}
/**
* Getter method for property <tt>to</tt>.
*
* @return property value of to
*/
public String getTo() {
return to;
}
/**
* Setter method for property <tt>to</tt>.
*
* @param to value to be assigned to property to
*/
public void setTo(String to) {
this.to = to;
}
/**
* Getter method for property <tt>content</tt>.
*
* @return property value of content
*/
public String getContent() {
return content;
}
/**
* Setter method for property <tt>content</tt>.
*
* @param content value to be assigned to property content
*/
public void setContent(String content) {
this.content = content;
}
}
/**
*
* @author zhangwei_david
* @version $Id: ProducerImpl.java, v 0.1 2015年1月31日 下午8:25:36 zhangwei_david Exp $
*/
@Component
public class ProducerImpl implements Producer {
@Autowired
private JmsTemplate jmsTemplate;
/**
*/
@Transactional
public void send(Mail mail) {
System.out.println("sende->" + mail);
jmsTemplate.convertAndSend(mail);
}
}
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:aop="http://www.springframework.org/schema/aop"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/shcema/jms
http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-3.0.xsd
http://www.springframework.org/schema/aop
http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
">
<aop:aspectj-autoproxy />
<context:annotation-config />
<context:component-scan base-package="com.cathy.demo.jms.*" />
<!-- connectionFactory -->
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
<!-- mailMessage converter -->
<bean id="mailMessageConverter" class="com.cathy.demo.jms.convert.MailMessageConverter"/>
<!-- jmsTemplate -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="defaultDestination" ref="topic"/>
<property name="receiveTimeout" value="60000"/>
<property name="pubSubDomain" value="true"/>
<property name="sessionTransacted" value="true"/>
<property name="messageConverter" ref="mailMessageConverter"/>
</bean>
<!--
<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="message.queue"/>
</bean>
-->
<!-- 主题 -->
<bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="notifyTopic"/>
</bean>
<bean id="defaultMessageListener" class="com.cathy.demo.jms.listener.MailMessageListener">
<property name="messageConverter" ref="mailMessageConverter"/>
</bean>
<!-- 消息接收监听器用于异步接收消息-->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="topic"/>
<property name="sessionTransacted" value="true"/>
<property name="messageListener" ref="defaultMessageListener"/>
</bean>
</beans>
/**
*
* @author zhangwei_david
* @version $Id: Sender.java, v 0.1 2015年1月31日 下午8:47:18 zhangwei_david Exp $
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath*:META-INF/spring/jms-beans.xml")
public class SenderAndReciver {
@Autowired
private Producer producer;
@Test
public void testSend() {
Mail mail = new Mail();
mail.setMailId("testId");
mail.setTo("david");
mail.setFrom("cathy");
mail.setContent("Hello");
producer.send(mail);
}
}
测试的结果是：
log4j:WARN Please initialize the log4j system properly. log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info. sende->Mail[mailId=testId,from=cathy,to=david,content=Hello] on message:Mail[mailId=testId,from=cathy,to=david,content=Hello]
MessageListenerAdapter
消息监听器适配器代理消息的处理目标 通过反射监听方法，具有灵活的消息类型转换。允许监听器的方法来对邮件内容类型进行操作，完全独立于JMS API
定义一个消息处理目标方法
/**
*
* @author zhangwei_david
* @version $Id: ListenerDelegate.java, v 0.1 2015年2月3日 下午2:32:11 zhangwei_david Exp $
*/
public class ListenerDelegate {
public void handleMessage(@SuppressWarnings("rawtypes") Map map) {
System.out.println("receive MapMessage->" + map);
}
}
<bean id="defaultListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="listenerDelegate"/>
</bean>
<!-- 消息接收监听器用于异步接收消息-->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<property name="destination" ref="topic"/>
<property name="sessionAcknowledgeModeName" value="CLIENT_ACKNOWLEDGE"/>
<property name="messageListener" ref="defaultListener"/>
<property name="sessionTransacted" value="true"/>
</bean>
执行测试方法的结果是：
sende->Mail[mailId=testId,from=cathy,to=david,content=Hello]
on message:Mail[mailId=testId,from=cathy,to=david,content=Hello]
receive MapMessage->{mailId=testId, from=cathy, to=david, content=Hello}
那么在MessageListenerAdapter中又是如何处理消息的呢？public void onMessage(Message message, Session session) throws JMSException {
// 获取代理方法
Object delegate = getDelegate();
// 如果代理方法不是当前类
if (delegate != this) {
// 如果代理方法是实现SessionAwareMessageListener接口的，直接将消息和session交给代理方法处理
if (delegate instanceof SessionAwareMessageListener) {
if (session != null) {
((SessionAwareMessageListener) delegate).onMessage(message, session);
return;
}
// session为null,且该代理对象未实现MessageListener接口则抛出一个异常
else if (!(delegate instanceof MessageListener)) {
throw new javax.jms.IllegalStateException("MessageListenerAdapter cannot handle a " +
"SessionAwareMessageListener delegate if it hasn't been invoked with a Session itself");
}
}
// 如果代理对象实现了MessageListener接口则将消息交给代理对象处理
if (delegate instanceof MessageListener) {
((MessageListener) delegate).onMessage(message);
return;
}
}
//将消息体转换为一个对象
Object convertedMessage = extractMessage(message);
// 获取这个消息的处理方法
String methodName = getListenerMethodName(message, convertedMessage);
if (methodName == null) {
throw new javax.jms.IllegalStateException("No default listener method specified: " +
"Either specify a non-null value for the 'defaultListenerMethod' property or " +
"override the 'getListenerMethodName' method.");
}
// 反射调用处理方法处理消息
Object[] listenerArguments = buildListenerArguments(convertedMessage);
Object result = invokeListenerMethod(methodName, listenerArguments);
if (result != null) {
handleResult(result, message, session);
}
else {
logger.trace("No result object given - no result to handle");
}
}