JMS以及JMS使用方式

1. 简介

　　　　

　　消息域的统一

　　　　JMS规范有两个主要的版本，1.0.2和1.1。

　　　　JMS1.0.2定义了两种消息域，点对点（队列）和发布/订阅（主题）。JMS 1.0.2的API为每个消息域提供了一个平行的类层次结构。导致客户端应用只能使用特定消息域的JMS API。JMS 1.1引进了统一消息域的概念使这两种消息域之间功能和客户端API的差别尽可能小。举个已消除的功能差异的例子，如果你使用的是JMS 1.1的消息供应者，你可以使用同一个Session事务性地在一个域消耗了一个消息后并且在另一个域中产生一个消息。

JMS 1.1的规范发布于2002年4月，并且在2003年11月成为J2EE 1.4的一个组成部分，结果，现在大多数使用的应用服务器只支持JMS 1.0.2的规范.

　　Spring提供了一个用于简化JMS API使用的抽象框架，并且对用户屏蔽了JMS API中1.0.2和1.1版本的差异。

　　　　JMS的功能大致上分为两块，叫做消息制造和消息消耗。JmsTemplate用于制造消息和同步消息接收。和Java EE的事件驱动bean风格类似，对于异步接收消息，Spring提供了一些消息侦听容器来创建消息驱动的POJO（MDP）。

　　　　org.springframework.jms.core包提供使用JMS的核心功能。 就象为JDBC提供的JdbcTemplate一样，它提供了JMS模板类来处理资源的创建和释放以简化JMS的使用。Spring模板类的公共设计原则就是通过提供助手方法去执行公共的操作，并将实际的处理任务委派到用户实现的回调接口上，从而完成更复杂的操作。 JMS模板也遵循这样的设计原则。这些类提供众多便利的方法来发送消息、同步接收消息、 使用户可以接触到JMS session和消息产生者。

　　　　org.springframework.jms.support包提供JMSException的转换功能。它将受控的 JMSException异常层次转换到一个对应的非受控异常层次。任何受控javax.jms.JMSException异常的子类都被包装在非受控UncategorizedJmsException异常里。

　　　　org.springframework.jms.support.converter 包提供一个MessageConverter用来抽象Java对象和JMS消息之间的转换操作。

　　　　包org.springframework.jms.support.destination为管理JMS目的地提供多种策略，例如为存储在JNDI中的目的地提供一个服务定位器。

　　　　最后，org.springframework.jms.connection包提供一个适合在独立应用中使用的 ConnectionFactory的实现。它还为JMS提供了一个Spring的PlatformTransactionManager的实现(现在叫做JmsTransactionManager)。 这样可以把JMS作为一个事务资源无缝地集成到Spring的事务管理机制中去。

　　

2. 使用Spring JMS

　

　　2.1. JmsTemplate

　　　　

　　　　JmsTemplate类有两个实现方式。JmsTemplate类使用JMS 1.1的API， 而子类JmsTemplate102使用了JMS 1.0.2的API。

使用JmsTemplate的代码只需要实现规范中定义的回调接口。 MessageCreator回调接口通过JmsTemplate中调用代码提供的Session来创建一条消息。 然而，为了允许更复杂的JMS API应用，回调接口SessionCallback为用户提供JMS session，并且回调接口ProducerCallback将Session和MessageProducer对显露给用户。

　　　　JMS API有两种发送方法，一种采用发送模式、优先级和存活时间作为服务质量（QOS）参数，另一种使用无需QOS参数的缺省值方法。由于在JmsTemplate中有许多种发送方法，QOS参数通过bean的属性方式进行设置,从而避免在多种发送方法中重复。同样，使用setReceiveTimeout属性值来设置同步接收调用的超时值。

　　　　某些JMS供应者允许通过ConnectionFactory的配置来设置缺省的QOS值。这样在调用MessageProducer的发送方法send(Destination destination, Message message)时会使用那些不同的QOS缺省值，而不是JMS规范中定义的值。所以，为了提供对QOS值的一致管理，JmsTemplate必须通过设置布尔值属性isExplicitQosEnabled为true，使它能够使用自己的QOS值。

　　

2.2. 连接工厂

 

　　　　JmsTemplate需要一个对ConnectionFactory的引用。ConnectionFactory是JMS规范的一部分，并且是使用JMS的入口。客户端应用通常用它作工厂配合JMS提供者去创建连接，并封装许多和供应商相关的配置参数，例如SSL的配置选项。

当在EJB里使用JMS时，供应商会提供JMS接口的实现，这样们可以参与声明式事务管理并 提供连接池和会话池。为了使用这个JMS实现，Java EE容器通常要求你在EJB或servlet部署描述符中声明一个JMS连接工厂做为 resource-ref。为确保可以在EJB内使用JmsTemplate的这些特性， 客户应用应当确保它引用了被管理的ConnectionFactory实现。

　　　　Spring提供了一个ConnectionFactory接口的实现，SingleConnectionFactory，它将在所有的createConnection调用中返回一个相同的Connection，并忽略所有对close的调用。这在测试和独立环境中相当有用，因为多个JmsTemplate调用可以使用同一个连接以跨越多个事务。SingleConnectionFactory通常引用一个来自JNDI的标准ConnectionFactory。

2.3. (消息)目的地管理

 

　　　　和连接工厂一样，目的地是可以在JNDI中存储和获取的JMS管理的对象。配置一个Spring应用上下文时，可以使用JNDI工厂类JndiObjectFactoryBean把对你对象的引用依赖注入到JMS目的地中。然而，如果在应用中有大量的目的地，或者JMS供应商提供了特有的高级目的地管理特性，这个策略常常显得很麻烦。创建动态目的地或支持目的地的命名空间层次就是这种高级目的地管理的例子。JmsTemplate将目的地名称到JMS目的地对象的解析委派给DestinationResolver接口的一个实现。JndiDestinationResolver是JmsTemplate 使用的默认实现，并且提供动态目的地解析。同时JndiDestinationResolver作为JNDI中的目的地服务定位器，还可选择回退去使用DynamicDestinationResolver中的行为。

　　　　经常见到一个JMS应用中使用的目的地在运行时才知道，因此，当部署一个应用时，它不能用可管理的方式创建。这是经常发生的，因为在互相作用的系统组件间有些共享应用逻辑会在运行的时候按照共同的命名规范创建消息目的地。虽然动态创建目的地不是JMS规范的一部分，但是大多数供应商已经提供了这个功能。 用户为动态创建的目的地定义和临时目的地不同的名字，并且通常不被注册到JNDI中。不同供应商创建动态消息目的地所使用的API差异很大，因为和目的地相关的属性是供应商特有的。然而，有时由供应商会作出一个简单的实现选择-忽略JMS规范中的警告，使用TopicSession的方法createTopic(String topicName)或者QueueSession的方法createQueue(String queueName)来创建一个带默认值属性的新目的地。依赖于供应商的实现，DynamicDestinationResolver也可能创建一个物理上的目的地，而不只是一个解析。

　　　　布尔属性pubSubDomain用来配置JmsTemplate使用什么样的JMS域。这个属性的默认值是false，使用点到点的域，也就是队列。在1.0.2的实现中，这个属性值用来决定JmsTemplate将消息发送到一个Queue还是一个Topic。这个标志在1.1的实现中对发送操作没有影响。然而，在这两个JMS版本中，这个属性决定了通过接口DestinationResolver的实现来决定如何解析动态消息目的地。

　　　　你还可以通过属性defaultDestination配置一个带有默认目的地的JmsTemplate。不指明目的地的发送和接受操作将使用该默认目的地。

2.4. 消息侦听容器

 

　　　　在EJB世界里，JMS消息最常用的功能之一是用于实现消息驱动bean(MDBs)。Spring提供了一个方法来创建消息驱动的POJO(MDPs),并且不会把用户绑定在某个EJB容器上。（关于Spring的MDP支持的细节请参考标题为Section 19.4.2, “异步接收 - 消息驱动的POJOs”的节）

　　　　通常用AbstractMessageListenerContainer的一个子类从JMS消息队列接收消息并驱动被注射进来的MDP。AbstractMessageListenerContainer负责消息接收的多线程处理并分发到各MDP中。一个消息侦听容器是MDP和消息提供者之间的一个中介，用来处理消息接收的注册，事务管理的参与，资源获取和释放，异常转换等等。这使得应用开发人员可以专注于开发和接收消息(可能的响应)相关的(复杂)业务逻辑，把和JMS基础框架有关的样板化的部分委托给框架处理。

Spring提供了三种AbstractMessageListenerContainer的子类，每种各有其特点。

2.4.1. SimpleMessageListenerContainer

 

　　　　这个消息侦听容器是三种中最简单的。它在启动时创建固定数量的JMS session并在容器的整个生命周期中使用它们。这个类不能动态的适应运行时的要求或参与消息接收的事务处理。然而它对JMS提供者的要求也最低。它只需要简单的JMS API。

2.4.2. DefaultMessageListenerContainer

 

　　　　这个消息侦听器使用的最多。和SimpleMessageListenerContainer一样，这个子类不能动态适应运行时侯的要求。然而，它可以参与事务管理。每个收到的消息都注册到一个XA事务中(如果配置过)，这样就可以利用XA事务语义的优势了。这个类在对JMS提供者的低要求和提供包括事务参于等的强大功能上取得了很好的平衡。

2.4.3. ServerSessionMessageListenerContainer

 

　　　　这个子类是三者中最强大的。它利用JMS ServerSessionPool SPI允许对JMS session进行动态管理。它也支持事务。使用这种消息侦听器可以获得强大的运行时调优功能，但是对使用到的JMS提供者有很高的要求(ServerSessionPool SPI)。如果不需要运行时的性能调整，请使用DefaultMessageListenerContainer或SimpleMessageListenerContainer。

2.5. 事务管理

 

　　　　Spring提供了一个JmsTransactionManager为单个JMSConnectionFactory管理事务。这将允许JMS应用利用Chapter 9, 事务管理中描述的Spring的事务管理功能。JmsTransactionManager从指定的ConnectionFactory绑定了一个Connection/Session对到线程上。然而，在Java EE环境中，SingleConnectionFactory将把连接和session放到缓冲池中，所以绑定到线程的实例将依赖越缓冲池的行为。在标准环境下，使用Spring的SingleConnectionFactory将使得和每个事务相关的JMS连接有自己的session。JmsTemplate也可以和JtaTransactionManager以及具有XA能力的JMS ConnectionFactory一起使用来提供分布式交易。

　　　　当使用JMS API从一个连接中创建session时，在受管理的和非受管理的事务环境下重用代码会可能会让人迷惑。这是因为JMS API只有一个工厂方法来创建session并且它需要用于事务和模式确认的值。在受管理的环境下，由事务结构环境负责设置这些值，这样在供应商包装的JMS连接中可以忽略这些值。当在一个非管理性的环境中使用JmsTemplate时，你可以通过使用属性SessionTransacted和SessionAcknowledgeMode来指定这些值。当配合 JmsTemplate中使用PlatformTransactionManager时，模板将一直被赋予一个事务性JMS的 Session。

3. 发送一条消息

　　　　JmsTemplate包含许多方便的方法来发送消息。有些发送方法可以使用 javax.jms.Destination对象指定目的地，也可以使用字符串在JNDI中查找目的地。没有目的地参数的发送方法使用默认的目的地。这里有个例子使用1.0.2版的JMS实现发送消息到一个队列。

import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.Session;

import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.JmsTemplate102;

public class JmsQueueSender {

  private JmsTemplate jmsTemplate;
  private Queue queue;

  public void setConnectionFactory(ConnectionFactory cf) {
    this.jmsTemplate = new JmsTemplate102(cf, false);
  }

  public void setQueue(Queue queue) {
    this.queue = queue;
  }

  public void simpleSend() {
    this.jmsTemplate.send(this.queue, new MessageCreator() {
      public Message createMessage(Session session) throws JMSException {
        return session.createTextMessage("hello queue world");
      }
    });
  }
}

　　　　这个例子使用MessageCreator回调接口从提供的Session对象中创建一个文本消息，并且通过一个ConnectionFactory的引用和指定消息域的布尔值来创建JmsTemplate。提供了一个无参数的构造方法和connectionFactory / queuebean属性并可用于创建实例（使用一个BeanFactory或者普通Java 代码code）。或者考虑从Spring的基类JmsGatewaySupport，它对JMS配置具有内置的bean属性，继承一个类。

　　　　当在应用上下文中配置JMS 1.0.2时，重要的是记得设定布尔属性pubSubDomain的值以指明你是要发送到队列还是主题。

　　　　方法send(String destinationName, MessageCreator creator)让你利用目的地的字符串名字发送消息。如果这个名字在JNDI中注册，你应当将模板中的destinationResolver属性设置为JndiDestinationResolver的一个实例。

　　　　如果你创建JmsTemplate并指定一个默认的目的地，send(MessageCreator c)发送消息到这个目的地。

3.1. 使用消息转换器

　　

　　　　为便于发送领域模型对象，JmsTemplate有多种以一个Java对象为参数并做为消息数据内容的发送方法。JmsTemplate里可重载的方法convertAndSend和receiveAndConvert将转换的过程委托给接口MessageConverter的一个实例。这个接口定义了一个简单的合约用来在Java对象和JMS消息间进行转换。缺省的实现SimpleMessageConverter支持String和TextMessage，byte[]和BytesMesssage,以及java.util.Map和MapMessage之间的转换。使用转换器，可以使你和你的应用关注于通过JMS接收和发送的业务对象而不用操心它是具体如何表达成JMS消息的。

　　　　目前的沙箱模型包括一个MapMessageConverter，它使用反射转换JavaBean和MapMessage。其他流行可选的实现方式包括使用已存在的XML编组的包，例如JAXB,Castor, XMLBeans, 或XStream的转换器来创建一个表示对象的TextMessage。

　　　　为方便那些不能以通用方式封装在转换类里的消息属性，消息头和消息体的设置，通过MessagePostProcessor接口你可以在消息被转换后并且在发送前访问该消息。下例展示了如何在java.util.Map已经转换成一个消息后更改消息头和属性。

public void sendWithConversion() {
  Map m = new HashMap();
  m.put("Name", "Mark");
  m.put("Age", new Integer(47));
  jmsTemplate.convertAndSend("testQueue", m, new MessagePostProcessor() {
    public Message postProcessMessage(Message message) throws JMSException {
      message.setIntProperty("AccountID", 1234);
      message.setJMSCorrelationID("123-00001");
      return message;
    }
  });
}

This results in a message of the form:

这将产生一个如下的消息格式:

MapMessage={ 
  Header={ 
    ... standard headers ...
    CorrelationID={123-00001} 
  } 
  Properties={ 
    AccountID={Integer:1234}
  } 
  Fields={ 
    Name={String:Mark} 
    Age={Integer:47} 
  } 
}

3.2. SessionCallback 和ProducerCallback

　　　　虽然send操作适用于许多常见的使用场景，但是有时你需要在一个JMS Session或者MessageProducer上执行多个操作。接口SessionCallback和ProducerCallback分别提供了JMS Session和Session / MessageProducer对。JmsTemplate上的execute()方法执行这些回调方法。

4. 接收消息

　　

4.1. 同步接收

 

　　　　虽然JMS一般都和异步处理相关，但它也可以同步的方式使用消息。可重载的receive(..)方法提供了这种功能。在同步接收中，接收线程被阻塞直至获得一个消息，有可能出现线程被无限阻塞的危险情况。属性receiveTimeout指定了接收器可等待消息的延时时间。

4.2. 异步接收 - 消息驱动的POJOs

　　　　类似于EJB世界里流行的消息驱动bean(MDB)，消息驱动POJO(MDP)作为JMS消息的接收器。MDP的一个约束(但也请看下面的有关javax.jms.MessageListener类的讨论)是它必须实现javax.jms.MessageListener接口。另外当你的POJO将以多线程的方式接收消息时必须确保你的代码是线程-安全的。

　　　　以下是MDP的一个简单实现:

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class ExampleListener implements MessageListener {

  public void onMessage(Message message) {
    if (message instanceof TextMessage) {
      try {
        System.out.println(((TextMessage) message).getText());
      } catch (JMSException ex) {
        throw new RuntimeException(ex);
      }
    } else {
      throw new IllegalArgumentException("Message must be of type TextMessage");
    }
  }
}

一旦你实现了MessageListener后就可以创建一个消息侦听容器。

请看下面例子是如何定义和配置一个随Sping发行的消息侦听容器的(这个例子用DefaultMessageListenerContainer)

<!-- this is the Message Driven POJO (MDP) -->
<bean id="messageListener" class="jmsexample.ExampleListener" />

<!-- and this is the attendant message listener container -->
<bean id="listenerContainer"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="concurrentConsumers" value="5"/>
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="destination" />
  <property name="messageListener" ref="messageListener" />
</bean>

关于各个消息侦听容器实现的特色请参阅相关的Spring Javadoc文档。

4.3. SessionAwareMessageListener 接口

　　　　SessionAwareMessageListener接口是一个Spring专门用来提供类似于JMS MessageListener的接口，也提供了从接收Message来访问JMS Session的消息处理方法。

package org.springframework.jms.listener;

public interface SessionAwareMessageListener {

    void onMessage(Message message, Session session) throws JMSException;
}

　　　　如果你希望你的MDP可以响应所有接收到的消息(使用onMessage(Message, Session)方法提供的Session)那么你可以选择让你的MDP实现这个接口(优先于标准的JMS MessageListener接口)。所有随Spring发行的支持MDP的消息侦听容器都支持MessageListener或SessionAwareMessageListener接口的实现。要注意的是实现了SessionAwareMessageListener接口的类通过接口和Spring有了耦合。是否选择使用它完全取决于开发者或架构师。

请注意SessionAwareMessageListener接口的'onMessage(..)'方法会抛出JMSException异常。和标准JMS MessageListener接口相反，当使用SessionAwareMessageListener接口时，客户端代码负责处理任何抛出的异常。

4.4. MessageListenerAdapter

　　

　　　　MessageListenerAdapter类是Spring的异步支持消息类中的不变类(final class)：简而言之，它允许你几乎将任意一个类做为MDP显露出来(当然有某些限制)。

　　　　如果你使用JMS 1.0.2 API，你将使用和MessageListenerAdapter一样功能的类MessageListenerAdapter102。

　　　　考虑如下接口定义。注意虽然这个接口既不是从MessageListener也不是从SessionAwareMessageListener继承来得，但通过MessageListenerAdapter类依然可以当作一个MDP来使用。同时也请注意各种消息处理方法是如何根据他们可以接收并处理消息的内容来进行强类型匹配的。

　　

public interface MessageDelegate {

    void handleMessage(String message);

    void handleMessage(Map message);

    void handleMessage(byte[] message);

    void handleMessage(Serializable message);
}

public class DefaultMessageDelegate implements MessageDelegate {
    // implementation elided for clarity...
}

特别请注意，上面的MessageDelegate接口(上文中DefaultMessageDelegate类)的实现完全不依赖于JMS。它是一个真正的POJO，我们可以通过如下配置把它设置成MDP。

<!-- this is the Message Driven POJO (MDP) -->
<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    <constructor-arg>
        <bean class="jmsexample.DefaultMessageDelegate"/>
    </constructor-arg>
</bean>

<!-- and this is the attendant message listener container... -->
<bean id="listenerContainer"
  class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="concurrentConsumers" value="5"/>
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="destination" />
  <property name="messageListener" ref="messageListener" />
</bean>

下面是另外一个只能处理接收JMSTextMessage消息的MDP示例。注意消息处理方法是如何实际调用'receive'(在MessageListenerAdapter中默认的消息处理方法的名字是'handleMessage')的，但是它是可配置的(你下面就将看到)。注意'receive(..)'方法是如何使用强制类型来只接收和处理JMS TextMessage消息的。

public interface TextMessageDelegate {

    void receive(TextMessage message);
}

public class DefaultTextMessageDelegate implements TextMessageDelegate {
    // implementation elided for clarity...
}

辅助的MessageListenerAdapter类配置文件类似如下：

<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    <constructor-arg>
        <bean class="jmsexample.DefaultTextMessageDelegate"/>
    </constructor-arg>
    <property name="defaultListenerMethod" value="receive"/>
    <!-- we don't want automatic message context extraction -->
    <property name="messageConverter">
        <null/>
    </property>
</bean>

请注意，如果上面的'messageListener'收到一个不是TextMessage类型的JMS Message，将会产生一个IllegalStateException异常(随之产生的其他异常只被捕获而不处理)。

MessageListenerAdapter还有一个功能就是如果处理方法返回一个非空值，它将自动返回一个响应消息。

请看下面的接口及其实现：

public interface ResponsiveTextMessageDelegate {

    // notice the return type...
    String receive(TextMessage message);
}

public class DefaultResponsiveTextMessageDelegate implements ResponsiveTextMessageDelegate {
    // implementation elided for clarity...
}

如果上面的DefaultResponsiveTextMessageDelegate和MessageListenerAdapter联合使用，那么任意从执行'receive(..)'方法返回的非空值都将(缺省情况下)转换成一个TextMessage。这个返回的TextMessage将被发送到原来的Message中JMS Reply-To属性定义的目的地(如果存在)，或者是MessageListenerAdapter设置(如果配置了)的缺省目的地；如果没有定义目的地，那么将产生一个InvalidDestinationException异常(此异常将不会只被捕获而不处理，它将沿着调用堆栈上传)。

4.5. 事务中的多方参与

　　

　　　　参与到事务中只需要一点微小的改动。你需要创建一个事务管理器，并且注册到一个可以参与事务的子类中(DefaultMessageListenerContainer或ServerSessionMessageListenerContainer)。

为了创建事务管理器，你需要创建一个JmsTransactionManager的实例并提供给它一个支持XA事务功能的连接工厂。

<bean id="transactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
  <property name="connectionFactory" ref="connectionFactory" />
</bean>

然后你只需要把它加入到我们先前的容器配置中。容器会处理其他的事情。

<bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="concurrentConsumers" value="5" />
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="destination" />
  <property name="messageListener" ref="messageListener" />
  <property name="transactionManager" ref="transactionManager" />
</bean>