JMS与Spring之二（用message listener container异步收发消息）

Posted 2023-01-02

本文主要参考Java Message Service（2nd）by Marc Ricbard。

在Spring框架中使用JMS传递消息有两种方式：JMS template和message listener container，前者用于同步收发消息，后者用于异步收发消息。

本文主要讲述用message listener container异步收发消息。 Spring中有三种方式配置异步消息监听器：实现javax.jms.MessageListener接口、实现Spring的SessionAwareMessageListener和捆绑一个标准POJO到Spring的MessageListenerAdapter类上。这三种方法在消息监听器类结构上不同，但都要用message listener container，message listener container类似JmsTemplate。

1.message listener container综述

message listener container绑定连接类工厂（connection factory）、JMS Destination、JNDI Destination解析器和message listener bean。Spring提供了两种message listener container：

DefaultMessageListenerContainer和SimpleMessageListenerContainer。两种message listener container都允许指定数量的并发监听线程，只有DefaultMessageListenerContainer可以在允许期间动态调整监听线程的数量。另外，DefaultMessageListenerContainer允许和XA Transactions的集成。对于使用本地事务管理器和不需要基于可变负载的线程、会话、连接调整的简单消息传递应用，使用SimpleMessageListenerContainer。对于使用外部事务管理器或XA事务的消息传递应用，使用DefaultMessageListenerContainer。

message listener container的配置类似JmsTemplate。使用JNDI访问连接工厂和JMS Destinations，或直接使用JMS提供者的native 连接工厂和JMS destination类。下面给出一个例子，使用了ActiveMQ的DefaultMessageListenerContainer。

Xml代码  
<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">
<property name="environment">
<props>
<prop key="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory
</prop>
<prop key="java.naming.provider.url">tcp://localhost:61616</prop>
<prop key="java.naming.security.principal">system</prop>
<prop key="java.naming.security.credentials">manager</prop>
</props>
</property>
</bean>
<bean id="jndiQueueConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiTemplate" ref="jndiTemplate"/>
<property name="jndiName" value="QueueCF"/>
</bean>
<bean id="queueConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="jndiQueueConnectionFactory"/>
<property name="sessionCacheSize" value="1"/>
</bean>
<bean id="queueDestination" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiTemplate" ref="jndiTemplate"/>
<property name="jndiName" value="queue1"/>
</bean>
<bean id="destinationResolver" class="org.springframework.jms.support.destination.JndiDestinationResolver">
<property name="jndiTemplate" ref="jndiTemplate"/>
<property name="cache" value="true"/>
<property name="fallbackToDynamicDestination" value="false"/>
</bean>
<bean id="messageListener1" class="com.jms.server.JMSReceiverAsync1" />
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="queueConnectionFactory"/>
<property name="destinationResolver" ref="destinationResolver"/>
<property name="destinationName" value="queue1"/>
<property name="messageListener" ref="messageListener1" />
</bean>

在这个配置中，messageListener1通过jmsContainer bean注入到DefaultMessageListenerContainer中，同时注入的还有destination名字、连接工厂、JNDI destination解析器。本例中，concurrentConsumers属性指示启动了三个并发监听线程。这个配置会应用到下面使用三种方法实现消息监听器的例子中。

2. 使用MessageListener接口

这是最简单的一种方式。如下图所示，POJO实现了MessageListener接口，并注入到DefaultMessageListenerContainer中。然后，DefaultMessageListenerContainer注入到CachingConnectFactory和JNDIDestinationResolver中。

实现MessageListener接口，需要重载onMessage方法。如下所示：

Java代码  
public void onMessage(Message message) 
try 
if (message instanceof TextMessage) 
System.out.println(((TextMessage) message).getText());
 else if (message instanceof MapMessage) 
MapMessage mmsg = (MapMessage) message;
System.out.println(mmsg.getLong("acctId") + ", " +
mmsg.getString("side") + ", " +
mmsg.getString("symbol") + ", " +
mmsg.getDouble("shares"));
 else 
throw new IllegalStateException("Message Type Not Supported");

 catch (JMSException e) 
e.printStackTrace();

public void onMessage(Message message) 
  try 
    if (message instanceof TextMessage) 
    System.out.println(((TextMessage) message).getText());
     else if (message instanceof MapMessage) 
    MapMessage mmsg = (MapMessage) message;

   

System.out.println(mmsg.getLong("acctId") + ", " +
    mmsg.getString("side") + ", " +
    mmsg.getString("symbol") + ", " +
    mmsg.getDouble("shares"));
     else 
    throw new IllegalStateException("Message Type Not Supported");
    
   catch (JMSException e) 
    e.printStackTrace();
  
  

3.使用SessionAwareMessageListener接口

SessionAwareMessageListener是Spring提供的javax.jms.MessageListener的扩展。类似地，它也包含了一个必须重载的onMessage方法。不同之处在于，SessionAwareMessageListener提供了对JMS Session的访问功能。配置方法和MessageListener相同。实现方法如下所示：

Java代码  
public void onMessage(Message message, Session session) throws JMSException 
if (message instanceof TextMessage) 
System.out.println(((TextMessage) message).getText());
 else if (message instanceof MapMessage) 
MapMessage mmsg = (MapMessage) message;
System.out.println(mmsg.getLong("acctId") + ", "
+ mmsg.getString("side") + ", " + mmsg.getString("symbol")
+ ", " + mmsg.getDouble("shares"));
 else 
throw new JMSException("Message type not supported");

long conf = 12345678;
MessageProducer sender = session.createProducer(message.getJMSReplyTo());
StreamMessage smsg = session.createStreamMessage();
smsg.setJMSCorrelationID(message.getJMSMessageID());
smsg.writeLong(conf);
sender.send(smsg);

public void onMessage(Message message, Session session) throws JMSException 
  if (message instanceof TextMessage) 
    System.out.println(((TextMessage) message).getText());
   else if (message instanceof MapMessage) 
    MapMessage mmsg = (MapMessage) message;
 
    System.out.println(mmsg.getLong("acctId") + ", "
      + mmsg.getString("side") + ", " + mmsg.getString("symbol")
      + ", " + mmsg.getDouble("shares"));
   else 
    throw new JMSException("Message type not supported");
  
 
  long conf = 12345678;
 
  MessageProducer sender = session.createProducer(message.getJMSReplyTo());
  StreamMessage smsg = session.createStreamMessage();
  
  smsg.setJMSCorrelationID(message.getJMSMessageID());
  smsg.writeLong(conf);
  sender.send(smsg);
  

4.使用MessageListenerAdapter

这个方法的特别之处在于POJO类不实现任何消息监听器接口，也不需要引用javax.jms.Message对象。如下图所示，POJO接收类注入Spring的MessageListenerAdapter，MessageListenerAdapter再注入DefaultMessageListenerContainer。

使用MessageListenerAdapter时，有多种方式安排POJO接收器方法。可以使用缺省的消息处理方法，也可以给监听类指定一个定制的方法。使用后者时，还需要指定是否使用一个消息转换器将收到的消息转换为相应的Java对象类型，或直接消费一个JMS Message。

（1） 缺省的消息处理方法

缺省情况下，MessageListenerAdapter会在POJO中根据收到的JMS消息类型寻找一个handleMessage方法。下面列出了使用自动消息转换时handleMessage方法的特征。

Java代码  
//receive a converted TextMessage body
public void handleMessage(String message) ...
//receive a converted BytesMessage body
public void handleMessage(byte[ ] message) ...
//receive a converted MapMessage body
public void handleMessage(Map message) ...
//receive a converted ObjectMessage body
public void handleMessage(Object message) ...
 //receive a converted TextMessage body
  public void handleMessage(String message) ...
 
  //receive a converted BytesMessage body
  public void handleMessage(byte[ ] message) ...
 
  //receive a converted MapMessage body
  public void handleMessage(Map message) ...
 
  //receive a converted ObjectMessage body
  public void handleMessage(Object message) ...

要使用缺省的消息监听处理方法，只需要通过constructor-arg属性（或delegate属性）将MDP注入MessageListenerAdapter bean。

Xml代码  
<bean id="messageListener3" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg>
<bean class="com.jms.server.JMSReceiverPOJO"/>
</constructor-arg>

</bean>
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="queueConnectionFactory"/>
<property name="destinationResolver" ref="destinationResolver"/>
<property name="destinationName" value="request"/>
<property name="messageListener" ref="messageListener3" />
</bean>

定义MDP（Message-driven POJO）时，只需要为需要处理的JMS message类型定义相应的handleMessage方法。如下面代码所示。

Java代码  
public class JMSReceiverPOJO 
public void handleMessage(Map message) 
System.out.println("Received Message A: " + message);

public void handleMessage(TradeData message) 
System.out.println("Received a tradedata object: ");

public void handleMessage(Trade2Data message) 
System.out.println("Received a trade2data object: ");

public class JMSReceiverPOJO 
 
  public void handleMessage(Map message) 
  System.out.println("Received Message A: " + message);
  
 
  public void handleMessage(TradeData message) 
  System.out.println("Received a tradedata object: ");
  
 
  public void handleMessage(Trade2Data message) 
  System.out.println("Received a trade2data object: ");
  

注意，JMSReceiverPOJO不需要引用任何JMS API，事实上，本例中的POJO甚至不知道它要用在消息传递上下文中。所有的消息传递架构都由MessageListenerAdapter和DefaultMessageListenerContainer处理。需要做的就是写相应的消息类型对应的handleMessage方法。如果收到的消息没有对应的handleMessExceptionage方法，会抛出NoSuchMethod。

上面的方法中，一个问题是消息转换只处理消息负载，头属性（header properties）和应用属性（application properties）没有访问。如果消息带有这些属性信息，需要告诉DefaultMessageListenerContainer不希望自动转换消息负载。要做到这点，需要将MessageListenerAdapter bean的messageConverter属性置为null。

Xml代码  
<bean id="messageListener3" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg>
<bean class="com.jms.server.JMSReceiverPOJO"/>
</constructor-arg>
<property name="messageConverter"><null/></property>
</bean>

这时，MessageListenerAdapter会缺省寻找下面的handleMessage方法。注意，上面的方法处理的是消息的body，此处的方法处理的消息全部。在方法内部可以访问头属性（header properties）和应用属性（application properties）。

Java代码  
// receive a JMS TextMessage
public void handleMessage(TextMessage message) ...
// receive a JMS MapMessage
public void handleMessage(MapMessage message) ...
  // receive a JMS TextMessage
  public void handleMessage(TextMessage message) ...
 
  // receive a JMS MapMessage
  public void handleMessage(MapMessage message) ...

（2） 定制消息处理方法

要使用定制的消息处理方法，需要将MessageListenerAdapter的defaultListenerMethod属性设置为自定义的方法。如下例所示。

Xml代码  
<bean id="messageListener3" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg>
<bean class="com.jms.server.JMSReceiverPOJO"/>
</constructor-arg>
<property name="defaultListenerMethod" value="processRequest"/>
<property name="messageConverter"><null/></property>
</bean>

Java代码  
public void processRequest(Map message) 
System.out.println(message.get("acctId") + ", " +
message.get("side") + ", " +
message.get("symbol") + ", " +
message.get("shares"));

public void processRequest(Map message) 
  System.out.println(message.get("acctId") + ", " +
  message.get("side") + ", " +
  message.get("symbol") + ", " +
  message.get("shares"));
  

5. PS：可以使用QueueBrowser查看队列中没有发送出去的消息。