JMS

JMS简介

1.   JMS基本概念

JMS(Java Message Service)是访问企业消息系统的标准API,它便于消息系统中的Java应用程序进行消息交换,并且通过提供标准的产生、发送、接收消息的接口简化企业应用的开发。

JMS应用由以下几部分组成：

JMS provider ：是一个消息系统，它实现了JMS 接口并提供管理和控制的功能。

JMS clients ：是用Java语言写的一些程序和组件，它们产生和使用消息。

Messages ：是在JMS clients之间传递的消息的对象。

Administered objects ：是由使用JMS clients 的人生成的预选设置好的JMS 对象。有两种这样的对象：destinations和connection factories。

Connection Factory

创建Connection对象的工厂，针对两种不同的JMS消息模型，分别有QueueConnectionFactory（队列工厂）和TopicConnectionFactory（主题工厂）两种。可以通过JNDI来查找ConnectionFactory对象。

Destination

Destination的意思是消息生产者的消息发送目标或者说消息消费者的消息来源。对于消息生产者来说，它的Destination是某个队列（Queue）或某个主题（Topic）；对于消息消费者来说，它的Destination也是某个队列或主题（即消息来源）。所以，Destination实际上就是两种类型的对象：Queue、Topic。可以通过JNDI来查找Destination。

Connection：

Connection表示在客户端和JMS系统之间建立的链接（对TCP/IP socket的包装）。Connection可以产生一个或多个Session。跟ConnectionFactory一样，Connection也有两种类型：QueueConnection和TopicConnection。

Session：

Session是我们操作消息的接口。可以通过session创建生产者、消费者、消息等。Session提供了事务的功能。当我们需要使用session发送/接收多个消息时，可以将这些发送/接收动作放到一个事务中。同样，也分QueueSession和TopicSession。

消息生产者：

消息生产者由Session创建，并用于将消息发送到Destination。同样，消息生产者分两种类型：QueueSender和TopicPublisher。可以调用消息生产者的方法（send或publish方法）发送消息。

消息消费者：

消息消费者由Session创建，用于接收被发送到Destination的消息。两种类型：QueueReceiver和TopicSubscriber。可分别通过session的createReceiver(Queue)或createSubscriber(Topic)来创建。当然，也可以通过session的createDurableSubscriber方法来创建持久化的订阅者。

MessageListener：

消息监听器。如果注册了消息监听器，一旦消息到达，将自动调用监听器的onMessage方法。EJB中的MDB（Message-Driven Bean）就是一种MessageListener。

MDB介绍：

对客户端来说，message-driven bean就是异步消息的消费者。当消息到达之后，由容器负责调用MDB。客户端发送消息到destination，MDB作为一个MessageListener接收消息。

2.   JMS基本功能

JMS是用于和面向消息的中间件相互通信的应用程序接口。它既支持点对点(point-to-point)的域，又支持发布/订阅(publish/subscribe)类型的域，并且提供对下列类型的支持：经认可的消息传递,事务型消息的传递，一致性消息和具有持久性的订阅者消息传递。

2.1.  消息服务类型

1)   point-to-point (PTP)方式：点到点的模型。

消息由一个JMS客户机（发布者）发送到服务器上的一个目的地，即一个队列（queue）。而另一个JMS客户机（订阅者）则可以访问这个队列，并从该服务器获取这条消息。

point-to-point (PTP)方式有以下特点：

a)   每一个message只有一个使用者。

b)   一个message的sender和receiver没有时间上的依赖关系。无论sendere有没有在运行，Receiver都可提取message。

c)   Receiver完成对message处理这后，发出确认。

d)   当你所发出的每一个消息必须由一个使用者成功处理的情况下，使用 PTP messaging机制。

2)   publish/subscribe (pub/sub)方式：发布-订阅模型。

这里仍然是由一个JMS客户机将一条消息发布到服务器上的一个目的地上，但是这次这个目的地叫做一个主题（topic），可有多个订阅者去访问该消息。消息将一直维持在主题中，直到这个主题的所有订阅者都取走了该消息的一个副本。消息也包括了一个参数，用于定义了该消息的耐久性（它能够在服务器上等待订阅者多长时间）。

Pub/sub messaging有如下的特点：

a)   每一个message可以有多个使用者;

b)   Publishers和subscribers在时间上有依赖关系。一个订阅了某一个topic的客户，只能使用在它生成订阅之后发布的message, 并且subscriber必须一直保持活动状态。

c)   JMS API允许客户生成持久性的订阅，从而在某种程度上放宽了这种时间上的依赖关系，提高了灵活性处可靠性。

3)   Messaging的使用

Messaging本身是异步的，使message的使用者之间没有时间上的依赖关系。但是，JMS规范给出了更精确的定义，使Message可以以两种方式被使用:

a)Synchronously同步：subscriber或receiver可以通过调用receive方法实时地从destination上提取message。Receive方法在收到一个 message后结束，或当message 在一定的时间限制内没有收到时超时结束。

b)Asynchronously异步：客户可以为某一个使用者注册一个message listener。message listener和event listener很相似。当一个message到达了destination, JMS provider通过调用listener的onMessage方法将message传递过去，由onMessage方法负责处理message。

更详细的JMS规范可参考SUN相关文档。

3.   消息的同步异步接收

   消息的同步接收是指客户端主动去接收消息，JMS 客户端可以采用MessageConsumer 的receive方法去接收下一个消息。

   消息的异步接收是指当消息到达时，主动通知客户端。JMS 客户端可以通过注册一个实 现MessageListener 接口的对象到MessageConsumer，这样，每当消息到达时，JMS Provider 会调用MessageListener中的onMessage 方法。

4.    PTP模型

在PTP模型中，有下列概念：消息队列(Queue)、发送者(Sender)、接收者(Receiver)。每个消息都被发送者(Sender)发送到一个特定的队列(Queue)，接收者(Receiver)从队列(Queue)中获取消息。队列保留着消息，直到它们被消费或超时。每个消息只有一个消费者(Consumer)（即一旦被消费，消息就不再在消息队列中）。发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列，队列的存在使得消息的异步传输成为可能。接收者在成功接收消息之后需向队列应答成功，如果你希望发送的每个消息都应该被成功处理的话，那么你需要P2P模型。

和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提供工具管理队列的创建、删除。PTP 模型定义了客户端如何向队列发送消息，从队列接收消息，浏览队列中的消息。

   下面描述JMS PTP 模型中的主要概念和对象：

名称描述

Queue由JMS Provider 管理，队列由队列名识别，客户端可以通过JNDI 接口用队列名得到一个队列对象。

TemporaryQueue由QueueConnection 创建，而且只能由创建它的QueueConnection 使用。

QueueConnectionFactory客户端用QueueConnectionFactory 创建QueueConnection 对象。

QueueConnection一个到JMS PTP provider 的连接，客户端可以用QueueConnection 创建QueueSession 来发送和接收消息。

QueueSession提供一些方法创建QueueReceiver 、QueueSender、QueueBrowser 和TemporaryQueue。如果在QueueSession 关闭时，有一些消息已经被收到，但还没有被签收(acknowledged)，那么，当接收者下次连接到相同的队列时，这些消息还会被再次接收。

QueueReceiver客户端用QueueReceiver 接收队列中的消息，如果用户在QueueReceiver 中设定了消息选择条件，那么不符合条件的消息会留在队列中，不会被接收到。

QueueSender客户端用QueueSender 发送消息到队列。

QueueBrowser客户端可以QueueBrowser 浏览队列中的消息，但不会收走消息。

QueueRequestorJMS 提供QueueRequestor 类简化消息的收发过程。QueueRequestor 的构造函数有两个参数：QueueSession 和queue，QueueRequestor 通过创建一个临时队列来完成最终的收发消息请求。

可靠性(Reliability)队列可以长久地保存消息直到接收者收到消息。接收者不需要因为担心消息会丢失而时刻和队列保持激活的连接状态，充分体现了异步传输模式的优势。

5.    PUB/SUB模型

在Pub/Sub模型中，有下列概念： 主题（Topic）、发布者（Publisher）、订阅者（Subscriber）。多个发布者（Publisher）将消息发送到Topic，系统将这些消息传递给多个订阅者（Subscriber）。 每个消息可以有多个消费者 。发布者和订阅者之间有时间上的依赖性。针对某个主题(Topic)的订阅者，它必须创建一个订阅之后，才能消费发布者的消息，而且，为了消费消息，订阅者必须保持运行的状态。 当然，为了缓和这种严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。 如果你希望发送的消息可以不被做任何处理、或者被一个消费者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型。

主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscribe)从主题订阅消息。主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。

   下面描述JMS Pub/Sub 模型中的主要概念和对象：

名称描述

订阅(subscription)消息订阅分为非持久订阅(non-durable subscription)和持久订阅(durable subscrip-tion)，非持久订阅只有当客户端处于激活状态，也就是和JMS Provider 保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。持久订阅时，客户端向JMS 注册一个识别自己身份的ID，当这个客户端处于离线时，JMS Provider 会为这个ID 保存所有发送到主题的消息，当客户再次连接到JMS Provider时，会根据自己的ID 得到所有当自己处于离线时发送到主题的消息。

Topic主题由JMS Provider 管理，主题由主题名识别，客户端可以通过JNDI 接口用主题名得到一个主题对象。JMS 没有给出主题的组织和层次结构的定义，由JMS Provider 自己定义。

TemporaryTopic临时主题由TopicConnection 创建，而且只能由创建它的TopicConnection 使用。临时主题不能提供持久订阅功能。

TopicConnectionFactory客户端用TopicConnectionFactory 创建TopicConnection 对象。

TopicConnectionTopicConnection 是一个到JMS Pub/Sub provider 的连接，客户端可以用TopicConnection创建TopicSession 来发布和订阅消息。

TopicSessionTopicSession 提供一些方法创建TopicPublisher、TopicSubscriber、TemporaryTopic 。它还提供unsubscribe 方法取消消息的持久订阅。

TopicPublisher客户端用TopicPublisher 发布消息到主题。

TopicSubscriber客户端用TopicSubscriber 接收发布到主题上的消息。可以在TopicSubscriber 中设置消息过滤功能，这样，不符合要求的消息不会被接收。

Durable TopicSubscriber如果一个客户端需要持久订阅消息，可以使用Durable TopicSubscriber，TopSession 提供一个方法createDurableSubscriber创建Durable TopicSubscriber 对象。

恢复和重新派送(Recovery and Redelivery)非持久订阅状态下，不能恢复或重新派送一个未签收的消息。只有持久订阅才能恢复或重新派送一个未签收的消息。

TopicRequestorJMS 提供TopicRequestor 类简化消息的收发过程。TopicRequestor 的构造函数有两个参数：TopicSession 和topic。TopicRequestor 通过创建一个临时主题来完成最终的发布和接收消息请求。

可靠性(Reliability)当所有的消息必须被接收，则用持久订阅模式。当丢失消息能够被容忍，则用非持久订阅模式。

6.   JMS API编程模型

一个JMS应用由以下几个模块组成：

6.1.  Administered Objects

JMS应用的destinations和connection factories最后是通过管理而不是编程来使用，因为不同的provider使用他们的方法不一样。JMS 客户应该使用统一的接口得到这些objects，从而使用JMS应用可以运行在不同provider上，而不需要修改或修改很少。通常管理员在JNDI上设置administered objects, 然后JMS clients 在JNDI上look up这些对象。

a)   Connection Factories：

connection factory 是client用来生成与provider的connection的对象。connection factory封装了一套由管理员定义的connection configuration参数。每个connection factory 是一个QueueConnectionFactory 或 TopicConnectionFactory接口的实例。

在JMS 客户程序中, 通常先执行connection factory 的JNDI API lookup。

 如下例：

Context ctx = new InitialContext();

QueueConnectionFactory queueConnectionFactory =

(QueueConnectionFactory) ctx.lookup("QueueConnectionFactory");

TopicConnectionFactory topicConnectionFactory =

(TopicConnectionFactory) ctx.lookup("TopicConnectionFactory");

如果调用不带参数的InitialContext的lookup方法，就在当前classpath 找jndi.properties文件。

b)   Destinations：

Destination 是一个对象用于定义用户产生的messages 的去向或用户使用的messages 的来源。

在PTP里，destinations被称作queues, 可以用下面的命令来生成它们:

Queue queue = (Queue) Queue.create("queue");

在pub/sub里, destinations被称为topics, 可以用下面的J2EE SDK command 来生成它们:

Topic topic = (Topic) Topic.create("topic");

一个JMS应用可以同时使用多个queues 和/或topics。

除了lookup connection factory, 也常要lookup destination。例如:

Topic myTopic = (Topic) ctx.lookup("MyTopic");

Queue myQueue = (Queue) ctx.lookup("MyQueue");

6.2.  Connection

Connection封装了一个与JMS provider的虚拟连接。Connection表示在client和provider service、 daemon之间打开的TCP/IP socket。可以用connection 生成一个或多个sessions。

就像connection factories。 connections有两种方式：实现QueueConnection或TopicConnection接口。

例如, 当有一个QueueConnectionFactory 或TopicConnectionFactory对象, 可以用他们来创造一个connection:

QueueConnection queueConnection =

queueConnectionFactory.createQueueConnection();

TopicConnection topicConnection =

topicConnectionFactory.createTopicConnection();

注意：当应用程序完成后, 必须关闭你所创建的connections。否则JMS provider 无法释放资源。关闭了connection同时也关闭了sessions和message产生者和message使用者。

queueConnection.close();

topicConnection.close();

在使用messages前, 必须调用connection的start方法。如果要暂时停止传送message而不关闭connection, 可以调用stop方法。

connection factory 是client用来生成与provider的connection的对象。connection factory封装了一套由管理员定义的connection configuration参数。每个connection factory 是一个QueueConnectionFactory 或 TopicConnectionFactory接口的实例。

6.3.  Session

Session是单线程的context用于产生和使用messages。用Session创建 message producers、message consumers和messages。Session管理message listeners的执行顺序。

Ssession提供事务模式，用于将一系列的sends和receives动作组合在一个工作单元里。

Session被标记为事务模式的话，确认消息就通过确认和校正来自动地处理。如果session没有标记为事务模式，有三个用于消息确认的选项:

• AUTO_ACKNOWLEDGE session将自动地确认收到一则消息。

• CLIENT_ACKNOWLEDGE 客户端程序将确认收到一则消息，调用这则消息的确认方法。

• DUPS_OK_ACKNOWLEDGE 这个选项命令session“懒散的”确认消息传递，可以想到，这将导致消息提供者传递的一些复制消息可能会出错。这种确认的方式只应当用于消息消费程序可以容忍潜在的副本消息存在的情况。

Sessions, 就象connections, 也有两种方式：实现QueueSession或TopicSession接口。例如:

TopicSession topicSession =

topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

第一个参数表示sessiong不实现事务处理; 第二个参数表示当session成功收到messages后自动确认。

相同的，可以用QueueConnection对象创建QueueSession:

QueueSession queueSession =

queueConnection.createQueueSession(true, 0);

这里, 第一个参数表示session实现事务处理; 第二个参数表示当session成功收到messages后不自动确认。

6.4.  Message Producers—消息生产者

message producer是由session创建的一个对象，用于将messages传递到目的地。

PTP方式的message producer实现QueueSender接口。 pub/sub方式的message producer实现TopicPublisher接口。

例如：:

QueueSender queueSender = queueSession.createSender(myQueue);

TopicPublisher topicPublisher = topicSession.createPublisher(myTopic);

用null作为createSender或createPublisher的参数，可以创建一个不确定的producer。用不确定的producer, 可以等到真正send或publish message的时候才指定destination

当创建了一个message producer, 就可以用它来发送messages。例如:

queueSender.send(message);

topicPublisher.publish(message);

6.5.  Message Consumer—消息消费者

message consumer也是由session创建的一个对象，用于接收发送到目的地的消息。message consumer允许JMS client到JMS provider注册感兴趣的目的地。JMS provider管理messages从destination到注册了这个destination的consumers之间的传送。

PTP方式的message consumer实现QueueReceiver接口。pub/sub方式的message consumer实现TopicSubscriber接口。

例如：

QueueReceiver queueReceiver = queueSession.createReceiver(myQueue);

TopicSubscriber topicSubscriber = topicSession.createSubscriber(myTopic);

可以用TopicSession.createDurableSubscriber方法创建一个durable topic subscriber（异步消息订阅 ）。

当创建了一个message consumer, 它就是活动的，就可以用它接收messages。可以用QueueReceiver 或TopicSubscriber的close方法把message consumer变成非活动的。Message的传送在调用了connection的start方法后才开始。

不论是QueueReceiver或TopicSubscriber, 都可以用receive方法来同步consume message。 可以在调用start方法后的任何时间调用它:

queueConnection.start();

Message m = queueReceiver.receive();

topicConnection.start();

Message m = topicSubscriber.receive(1000); // time out after a second

异步consume message, 可以使用message listener。

6.5.1.   Message Listeners

message listener是一个对象，用作充当messages的异步事件处理器。它实现了MessageListener接口, 它只有一个方法：onMessage。 在onMessage方法内, 可以定义当收到一个message后做的事情。

用setMessageListener方法在某个QueueReceiver 或TopicSubscriber里注册message listener。例如：

TopicListener topicListener = new TopicListener();

topicSubscriber.setMessageListener(topicListener);

当注册了message listener, 调用QueueConnection或TopicConnection的方法来开始传送message。

当message开始传送, 当有message送来，message consumer自动调用message listener的 onMessage方法。onMessage方法只有一个Message类型的参数。

message listener并不对应特定的destination类型. 相同的listener可以从queue或topic上得到message, 这取决于listener是由QueueReceiver还是 TopicSubscriber对象设置的。然而message listener通常对应某一个message类型或格式, 如果要回应messages, message listener必须创建一个message producer。

onMessage方法应该处理所有的exceptions。

Session负责管理message listeners的执行顺序。任何时候，只有一个message listeners在运行。

6.5.2.   Message Selectors

如果你的消息应用程序需要过滤收到的messages, 可以用JMS API中的message selector来让message consumer定义它所感兴趣的messages。Message selectors负责过滤到JMS provider的message，而不是到应用程序的。

message selector是一个含有表达式的字符串。表达式的语法是SQL92 conditional expression syntax的一个子集。当创建message consumer时， createReceiver, createSubscriber, 和createDurableSubscriber方法都可以定义某个message selector作为参数。

message consumer只接收headers和properties与selector匹配的messages。message selector不能根据message body的内容进行选择。

6.6.  Message—消息组成

JMS 消息由以下几部分组成：消息头，属性，消息体。

6.6.1.   消息头(Header)

消息头包含消息的识别信息和路由信息，消息头包含一些标准的属性如：JMSDestination,JMSMessageID 等。

消息头由一下类定义：

JMSDestinationsend 或 publish 方法

JMSDeliveryModesend 或 publish 方法

JMSExpirationsend 或 publish 方法

JMSPrioritysend 或 publish 方法

JMSMessageIDsend 或 publish 方法

JMSTimestampsend 或 publish 方法

JMSCorrelationID客户

JMSReplyTo客户

JMSType客户

JMSRedeliveredJMS Provider

6.6.2.   属性(Properties)

除了消息头中定义好的标准属性外，JMS 提供一种机制增加新属性到消息头 中，这种新属性包含以下几种：

1. 应用需要用到的属性;

2. 消息头中原有的一些可选属性;

 JMS Provider 需要用到的属性。

标准的JMS 消息头包含以下属性：

JMSDestination消息发送的目的地

JMSDeliveryMode传递模式，有两种模式： PERSISTENT 和NON_PERSISTENT，PERSISTENT 表示该消息一定要被送到目的地，否则会导致应用错误。NON_PERSISTENT 表示偶然丢失该消息是被允许的，这两种模式使开发者可以在消息传递的可靠性和吞吐量之间找到平衡点。

JMSMessageID唯一识别每个消息的标识，由JMS Provider 产生。

JMSTimestamp一个消息被提交给JMS Provider 到消息被发出的时间。

JMSCorrelationID用来连接到另外一个消息，典型的应用是在回复消息中连接到原消息。

JMSReplyTo提供本消息回复消息的目的地址

JMSRedelivered如果一个客户端收到一个设置了JMSRedelivered 属性的消息，则表示可能该客户端曾经在早些时候收到过该消息，但并没有签收(acknowledged)。

JMSType消息类型的识别符。

JMSExpiration消息过期时间，等于QueueSender 的send 方法中的timeToLive 值或TopicPublisher 的publish 方法中的timeToLive 值加上发送时刻的GMT 时间值。如果timeToLive值等于零，则JMSExpiration 被设为零，表示该消息永不过期。如果发送后，在消息过期时间之后消息还没有被发送到目的地，则该消息被清除。

JMSPriority消息优先级，从0-9 十个级别，0-4 是普通消息，5-9 是加急消息。JMS 不要求JMS Provider 严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。

6.6.3.   消息体(Body)

JMS API 定义了5种消息体格式，也叫消息类型，你可以使用不同形式发送接收 数据并可以兼容现有的消息格式，下面描述这5种类型：

消息类型消息体

TextMessagejava.lang.String对象，如xml文件内容

MapMessage名/值对的集合，名是String对象，值类型可以是Java任何基本类型

BytesMessage字节流

StreamMessageJava中的输入输出流

ObjectMessageJava中的可序列化对象

Message没有消息体，只有消息头和属性

6.7.  Exception Handling

JMS API的方法的Exception的根类是JMSException。JMSException类包括如下子类:

IllegalStateException

InvalidClientIDException

InvalidDestinationException

InvalidSelectorException

JMSSecurityException

MessageEOFException

MessageFormatException

MessageNotReadableException

MessageNotWriteableException

ResourceAllocationException

TransactionInProgressException

TransactionRolledBackException