JMS应用

JMS应用

JMS（Java Message Service，Java消息服务）是一组Java应用程序接口(Java API)，它提供创建、发送、接收、读取消息的服务。由Sun公司和它的合作伙伴设计的JMS API定义了一组公共的应用程序接口和相应语法，使得Java程序能够和其他消息组件进行通信。
JMS是一种与厂商无关的 API，用来访问消息收发系统。它类似于 JDBC (Java Database Connectivity)：这里，JDBC 是可以用来访问许多不同关系数据库的 API，而 JMS 则提供同样与厂商无关的访问方法，以访问消息收发服务。许多厂商目前都支持 JMS，包括 IBM 的 MQSeries、BEA的 Weblogic JMS service和 Progress 的 SonicMQ，这只是几个例子。
JMS 使您能够通过消息收发服务（有时称为消息中介程序或路由器）从一个 JMS 客户机向另一个 JML 客户机发送消息。消息是 JMS 中的一种类型对象，由两部分组成：报头和消息主体。报头由路由信息以及有关该消息的元数据组成。消息主体则携带着应用程序的数据或有效负载。根据有效负载的类型来划分，可以将消息分为几种类型，它们分别携带：简单文本 (TextMessage)、可序列化的对象 (ObjectMessage)、属性集合 (MapMessage)、字节流 (BytesMessage)、原始值流 (StreamMessage)，还有无有效负载的消息 (Message)。

消息收发系统是异步的，也就是说，JMS 客户机可以发送消息而不必等待回应。比较可知，这完全不同于基于 RPC 的（基于远程过程的）系统，如 EJB 1.1、CORBA 和 Java RMI 的引用实现。在 RPC 中，客户机调用服务器上某个分布式对象的一个方法。在方法调用返回之前，该客户机被阻塞；该客户机在可以执行下一条指令之前，必须等待方法调用结束。在 JMS 中，客户机将消息发送给一个虚拟通道（主题或队列），而其它 JMS 客户机则预订或监听这个虚拟通道。当 JMS 客户机发送消息时，它并不等待回应。它执行发送操作，然后继续执行下一条指令。消息可能最终转发到一个或许多个客户机，这些客户机都不需要作出回应。

　　JMS的通用接口集合以异步方式发送或接收消息。异步方式接收消息显然是使用间断网络连接的客户机，诸如移动电话和PDA的最好的选择。另外， JMS采用一种宽松结合方式整合企业系统的方法，其主要的目的就是创建能够使用跨平台数据信息的、可移植的企业级应用程序，而把开发人力解放出来。

　　Java消息服务支持两种消息模型：Point-to-Point消息(P2P)和发布订阅消息（Publish Subscribe messaging，简称Pub/Sub）。JMS规范并不要求供应商同时支持这两种消息模型，但开发者应该熟悉这两种消息模型的优势与缺点。

　　P2P消息模型是在点对点之间传递消息时使用。如果应用程序开发者希望每一条消息都能够被处理，那么应该使用P2P消息模型。与Pub/Sub消息模型不同，P2P消息总是能够被传送到指定的位置。

　　Pub/Sub模型在一到多的消息广播时使用。如果一定程度的消息传递的不可靠性可以被接受的话，那么应用程序开发者也可以使用Pub/Sub消息模型。换句话说，它适用于所有的消息消费程序并不要求能够收到所有的信息或者消息消费程序并不想接收到任何消息的情况。

　　JMS通过允许创建持久订阅来简化时间相关性，即使消息预订者未激活也可以接收到消息。此外，使用持久订阅还可通过队列提供灵活性和可靠性，而仍然允许消息被发给许多的接收者。 Topic Subscriber topic Subscriber = topicSession.createDurableSubscriber(topic, subscriptionName); 　　Connection对象表示了到两种消息模型中的任一种的消息系统的连接。服务器端和客户机端对象要求管理创建的JMS连接的状态。连接是由Connection Factory创建的并且通过JNDI查寻定位。

//取得用于 P2P的 QueueConnectionFactory QueueConnectionFactory = queueConnectionFactory( ); Context messaging = new InitialContext( ); QueueConnectionFactory = (QueueConnectionFactory) Messaging.lookup(“QueueConnectionFactory”); //取得用于 pub/sub的 TopicConnectionFactory TopicConnectonFactory topicConnectionFactory; Context messaging = new InitialContext(); topicConnectionFactory = (TopicConnectionFactory) messaging.lookup(“TopicConnectionFactory”);

　注意：用于P2P的代码和用于PublishSubscribe的代码非常相似。

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

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

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

queueSession = queueConnection.createQueueSession(false, session.AUTO_ACKNOWLEDGE);//P2P topicSession = topicConnection.createTopicSession(false, session.AUTO_ACKNOWLEDGE); //Pub-Sub

　　注意：在本例中，一个session目的从连结中创建，非值指出session是non-transactional的，并且 session将自动地确认收到一则消息。

　　JMS现在有两种传递消息的方式。标记为NON_PERSISTENT的消息最多投递一次，而标记为PERSISTENT的消息将使用暂存后再转送的机理投递。如果一个JMS服务离线，那么持久性消息不会丢失但是得等到这个服务恢复联机时才会被传递。所以默认的消息传递方式是非持久性的。即使使用非持久性消息可能降低内务和需要的存储器，并且这种传递方式只有当你不需要接收所有的消息时才使用。

　　虽然 JMS规范并不需要JMS供应商实现消息的优先级路线，但是它需要递送加快的消息优先于普通级别的消息。JMS定义了从0到9的优先级路线级别，0是最低的优先级而9则是最高的。更特殊的是0到4是正常优先级的变化幅度，而5到9是加快的优先级的变化幅度。举例来说： topicPublisher.publish (message, DeliveryMode.PERSISTENT, 8, 10000); //Pub-Sub 或 queueSender.send(message, DeliveryMode.PERSISTENT, 8, 10000);//P2P 　　这个代码片断，有两种消息模型，映射递送方式是持久的，优先级为加快型，生存周期是10000 (以毫秒度量 )。如果生存周期设置为零，这则消息将永远不会过期。当消息需要时间限制否则将使其无效时，设置生存周期是有用的。

　　JMS定义了五种不同的消息正文格式，以及调用的消息类型，允许你发送并接收以一些不同形式的数据，提供现有消息格式的一些级别的兼容性。

　　· StreamMessage -- Java原始值的数据流

　　· MapMessage--一套名称-值对

　　· TextMessage--一个字符串对象

　　· ObjectMessage--一个序列化的 Java对象

　　· BytesMessage--一个未解释字节的数据流

　　JMS应用程序接口提供用于创建每种类型消息和设置荷载的方法例如，为了在一个队列创建并发送一个TextMessage实例，你可以使用下列语句：

TextMessage message = queueSession.createTextMessage(); message.setText(textMsg); 　　以异步方式接收消息，需要创建一个消息监听器然后注册一个或多个使用MessageConsumer的JMS MessageListener接口。会话(主题或队列)负责产生某些消息，这些消息被传送到使用onMessage方法的监听者那里。 import javax.jms.*; public class ExampleListener implements MessageListener { //把消息强制转化为TextMessage格式 public void onMessage(Message message) { TextMessage textMsg = null; // 打开并处理这段消息 } } 　　当我们创建QueueReceiver和TopicSubscriber时，我们传递消息选择器字符串： //P2P QueueReceiver QueueReceiver receiver; receiver = session.createReceiver(queue, selector); //Pub-Sub TopicSubscriber TopicSubscriber subscriber; subscriber = session.createSubscriber(topic, selector); 　　为了启动消息的交付，不论是Pub/Sub还是P2P，都需要调用start方法。 TopicConnection.start( ); //pub-sub QueueConnection.start( ); //P2P TopicConnection.start ( );// pub-sub QueueConnection.start ( );// P2P

　　当一条消息被捕捉时，这条消息做为一条必须被强制转化为适当消息类型的普通Message对象到达。这是一个被用来提取或打开消息内容的getter方法。下列代码片段使用StreamMessage类型。 private void unPackMessage (Message message) { String eName; String position; double rate; StreamMessage message; Message = session.createStreamMessage( ); //注意下面的代码必须按照我给出的顺序书写 message.writeString(eName); message.writeString(position); message.writeDouble(rate); //实现处理消息的必要的程序逻辑 }

　　停止消息的传递，无论是Pub/Sub还是P2P，都调用stop方法。 TopicConnection.start( ); //pub-sub QueueConnection.start( ); //P2P TopicConnection.start ( );// pub-sub QueueConnection.start ( );// P2P 　　其他的J2EE组件--servlet或EJB--可以当作消息生产者；然而，它们可能只能同步操作，这可能是因为它们的请求-应答的性质决定的。虽然XML目前还不是被支持的消息类型，发送一个XML文件和创建一条文本类型消息以及把XML文件添加到消息的有效负载都一样简单，都是以非专有的方式传送数据。值得注意的是，一些JMS供应厂商已经提供了可用的XML消息类型。但是使用非标准的消息类型可能会出现可移植性问题。 String reportData; //reportData内容为XML 文档 TextMessage message; message = session.createTextMessage(); message.setText (reportData);

　　消息驱动组件(MDB)是一个当消息到达时被容器调用的异步消息消费程序。和entity和session EJB不同，MDB没有本地和远程接口并且是匿名的；它们对于客户是不可见的。MDB是JMS系统的一部分，作为消费者实现服务器上的商业逻辑程序。 一个客户程序可能通过使用JNDI定位一个与MDB相关联的JMS。 例如：

Context initialContext = new InitialContext(); Queue reportInfoQueue = (javax.jms.Queue)initialContext.lookup (“java:comp/env/jms/reportInfoQueue”); 　　MDB是由Bean类和相应的XML部署描述符组成。 Bean 类实现MessageDriveBean 接口： import javax.ejb.*; import jms.Message.*; public interface MessageDriveBean { public void ejbCreate(); public void ejbRemove(); public void setMessageDrivenContext(MessageDrivenContext ctx); } 　　消息监听器接口： import javax.jms.*; public interface MessageListener { public void onMessage( ); }

　　部署描述符

＜!DOCTYPE ejb-jar PUBLIC "-//Sun Microsystems, Inc.//DTD Enterprise JavaBeans 2.0//EN" "http://java.sun.com/j2ee/dtds/ejb-jar_2_0.dtd"＞ ＜ejb-jar＞ ＜enterprise-beans＞ ＜message-driven＞ ＜ejb-name＞MDB＜/ejb-name＞ ＜ejb-class＞MDB＜/ejb-class＞ ＜transaction-type＞Container＜/transaction-type＞ ＜message-driven-destination＞ ＜jms-destination-type＞javax.jms.Queue＜/jms-destination-type＞ ＜/message-driven-destination＞ ＜security-identity＞ ＜run-as-specified-identity＞ ＜role-name＞everyone＜/role-name＞ ＜/run-as-specified-identity＞ ＜/security-identity＞ ＜/message-driven＞ ＜/enterprise-beans＞ ＜/ejb-jar＞

　　既然我们现在已经有了一些基本的JMS知识，那么我们可以使用JMS做什么呢？任何事情都可以。

　　例如，分别用于销售、库存、客户服务和账目处理的系统。这些部门之间的系统很可能已经存在了很长时间，这些处理要求把事务移动到系统中去，这并不是一个小的工作。这就是消息服务适用的地点。

　　当售货员完成销售的时候，一条消息被发给库存系统；一旦订单消息发送给收发货人员，就可以按照订单出货了。当订单成功地发货，系统将通知顾客服务和会计系统这个订单已经成功的交易了。所有对应的每个子系统都自动地根据收到的消息进行更新。

　　JMS一般都不是用来整合一个系统，而是整合许多可能参与消息驱动环境的系统。JMS是一个用于开发和集成企业应用程序的重要的工具。因为许多公司都有以前遗留下来的系统和新近开发的系统综合起来的系统，消息的使用是整合整个企业的重要的步骤。

　　

JMS接口描述

JMS 支持两种消息类型PTP 和Pub/Sub，分别称作：PTP Domain 和Pub/Sub Domain，这两种接口都继承统一的JMS Parent 接口，JMS 主要接口如下所示：


JMS Parent PTPDomain Pub/Sub Domain
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver,QueueBrowser TopicSubscriber


以下是对这些接口的简单描述：
ConnectionFactory ：连接工厂，JMS 用它创建连接
Connection ：JMS 客户端到JMS Provider 的连接
Destination ：消息的目的地
Session： 一个发送或接收消息的线程
MessageProducer： 由Session 对象创建的用来发送消息的对象
MessageConsumer： 由Session 对象创建的用来接收消息的对象



JMS消息模型

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

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


消息头 由谁设置
JMSDestination send 或 publish 方法
JMSDeliveryMode send 或 publish 方法
JMSExpiration send 或 publish 方法
JMSPriority send 或 publish 方法
JMSMessageID send 或 publish 方法
JMSTimestamp send 或 publish 方法
JMSCorrelationID 客户
JMSReplyTo 客户
JMSType 客户
JMSRedelivered JMS Provider


属性(Properties) - 除了消息头中定义好的标准属性外，JMS 提供一种机制增加新属性到消息头中，这种新属性包含以下几种：
1. 应用需要用到的属性;
2. 消息头中原有的一些可选属性;
3. 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 严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。

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


消息类型 消息体
TextMessage java.lang.String对象，如xml文件内容
MapMessage 名/值对的集合，名是String对象，值类型可以是Java任何基本类型
BytesMessage 字节流
StreamMessage Java中的输入输出流
ObjectMessage Java中的可序列化对象
Message 没有消息体，只有消息头和属性。


下例演示创建并发送一个TextMessage到一个队列：

TextMessage message = queueSession.createTextMessage(); 

message.setText(msg_text); // msg_text is a String 

queueSender.send(message);

下例演示接收消息并转换为合适的消息类型：

Message m = queueReceiver.receive(); 

if (m instanceof TextMessage) { 

 TextMessage message = (TextMessage) m; 

 System.out.println("Reading message: " + message.getText()); 

} else { 

 // Handle error