dev_liu
- 浏览: 112035 次
- 性别:
- 来自: 成都
-
最近访客 更多访客>>
文章分类
最新评论
-
dev_liu：
lipengyu2006 写道现在怎么样儿了啊 房子多少钱买 ...
最近比较心烦 -
lipengyu2006：
现在怎么样儿了啊 房子多少钱买的。
最近比较心烦 -
cynan168：
...
hibernate数据查询的几种方式 -
My_Choice：
joram中文文档 -
My_Choice：
非常感谢，太有用了，中文文档不好找啊，而且是这么详细的
joram中文文档
详细看评论
分享到：
评论
3 楼
My_Choice
2008-05-15
2 楼
My_Choice
2008-03-25
非常感谢，太有用了，中文文档不好找啊，而且是这么详细的
1 楼
dev_liu
2007-01-31
1 引言
1.1 编写目的
本文作为B2bi项目中开源产品JORAM的使用指导文档，旨在帮助项目组人员方便明了的进行JMS模块的详细设计和开发工作。本文档主要包含建设银行EAI平台B2Bi子系统中使用的开源JMS产品——JORAM的使用说明。
1.2 名词解释
B2Bi：
Business to Business integration （企业间集成）
JMS:
Java Message Service （Java消息服务）
JORAM:
ObjectWeb的Java开源项目
JNDI：
Java命名和目录接口
1.3 参考资料
《Joram-4.3-en.pdf》——JORAM使用手册（英文）
《Joram4_0_SAMPLES.pdf》——JORAM使用举例（英文）
《Joram4_0_ADMIN.pdf》——JORAM管理员手册（英文）
2 JMS简介
2.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。
2.2 JMS基本功能
JMS是用于和面向消息的中间件相互通信的应用程序接口。它既支持点对点(point-to-point)的域，又支持发布/订阅(publish/subscribe)类型的域，并且提供对下列类型的支持：经认可的消息传递,事务型消息的传递，一致性消息和具有持久性的订阅者支持。JMS还提供了另一种方式来对您的应用与旧的后台系统相集成。
2.3 消息服务类型
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相关文档。
2.4 JMS接口类图
2.5 JMS基本视图
1.JMS接口描述
JMS 支持两种消息类型PTP 和Pub/Sub，分别称作：PTP Domain 和Pub/Sub Domain，这两种接口都继承统一的JMS父接口，JMS 主要接口如下所示：
MS父接口 PTP Pub/Sub
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver,QueueBrowse r TopicSubscriber
ConnectionFactory ：连接工厂，JMS 用它创建连接
Connection ：JMS 客户端到JMS Provider 的连接
Destination ：消息的目的地
Session： 一个发送或接收消息的线程
MessageProducer： 由Session 对象创建的用来发送消息的对象
MessageConsumer： 由Session 对象创建的用来接收消息的对象
2.JMS消息模型
JMS 消息由以下几部分组成：消息头，属性，消息体。
2.1 消息头(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
2.2 属性(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 严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。
2.3 消息体(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
}
3. 消息的同步异步接收
消息的同步接收是指客户端主动去接收消息，JMS 客户端可以采用MessageConsumer 的receive方法去接收下一个消息。
消息的异步接收是指当消息到达时，主动通知客户端。JMS 客户端可以通过注册一个实 现MessageListener 接口的对象到MessageConsumer，这样，每当消息到达时，JMS Provider 会调用MessageListener中的onMessage 方法。
4. PTP模型
PTP(Point-to-Point)模型是基于队列的，发送方发消息到队列，接收方从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提 供工具管理队列的创建、删除。JMS 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 浏览队列中的消息，但不会收走消息。
QueueRequestor JMS 提供QueueRequestor 类简化消息的收发过程。QueueRequestor 的构造函数有两个参数：QueueSession 和queue，QueueRequestor 通过创建一个临时队列来完成最终的收发消息请求。
可靠性(Reliability) 队列可以长久地保存消息直到接收者收到消息。接收者不需要因为担心消息会丢失而时刻和队列保持激活的连接状态，充分体现了异步传输模式的优势。
5. PUB/SUB模型
JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作主题(topic)。
主题可以被认为是消息的传输中介，发布者(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 对象。
TopicConnection TopicConnection 是一个到JMS Pub/Sub provider 的连接，客户端可以用TopicConnection创建TopicSession 来发布和订阅消息。
TopicSession TopicSession 提供一些方法创建TopicPublisher、TopicSubscriber、TemporaryTopic 。它还提供unsubscribe 方法取消消息的持久订阅。
TopicPublisher 客户端用TopicPublisher 发布消息到主题。
TopicSubscriber 客户端用TopicSubscriber 接收发布到主题上的消息。可以在TopicSubscriber 中设置消息过滤功能，这样，不符合要求的消息不会被接收。
Durable TopicSubscriber 如果一个客户端需要持久订阅消息，可以使用Durable TopicSubscriber，TopSession 提供一个方法createDurableSubscriber创建Durable TopicSubscriber 对象。
恢复和重新派送(Recovery and Redelivery) 非持久订阅状态下，不能恢复或重新派送一个未签收的消息。只有持久订阅才能恢复或重新派送一个未签收的消息。
TopicRequestor JMS 提供TopicRequestor 类简化消息的收发过程。TopicRequestor 的构造函数有两个参数：TopicSession 和topic。TopicRequestor 通过创建一个临时主题来完成最终的发布和接收消息请求。
可靠性(Reliability) 当所有的消息必须被接收，则用持久订阅模式。当丢失消息能够被容忍，则用非持久订阅模式。
3 JMS API编程模型
一个JMS应用由以下几个模块组成：
3.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");
3.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接口的实例。
3.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后不自动确认。
3.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);
3.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。
a) 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在运行。
b) 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的内容进行选择。
3.6 Message—消息组成
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 没有消息体，只有消息头和属性。
JMS API为每一种messages都提供了一种create方法。例如：
TextMessage message = queueSession.createTextMessage();
message.setText(msg_text); // msg_text is a String
queueSender.send(message);
在使用者一端, 必须将收到的Message 按照适当的message类型处理。 例如:
Message m = queueReceiver.receive();
if (m instanceof TextMessage) {
TextMessage message = (TextMessage) m;
System.out.println("Reading message: " + message.getText());
} else {
// Handle error
}
3.7 Exception Handling
JMS API的方法的Exception的根类是JMSException。JMSException类包括如下子类:
IllegalStateException
InvalidClientIDException
InvalidDestinationException
InvalidSelectorException
JMSSecurityException
MessageEOFException
MessageFormatException
MessageNotReadableException
MessageNotWriteableException
ResourceAllocationException
TransactionInProgressException
TransactionRolledBackException
4 JORAM产品介绍
JORAM是法国电信研究院支持的objectweb.org的开源JMS消息中间件产品。objectweb是一个非常活跃的开源中间件团体。JORAM支持sun规范，实现JMS1.1。
4.1 Joram4.3包括
一个Message Server（消息服务器），提供Java的消息服务功能。
一个JNDI Server。
一个客户端类，可以访问Message Server上的消息。
一个图形化的管理界面，配置管理Joram平台。
4.2 系统要求
硬件要求：
2000年后的intel架构pc机
256M RAM 5G硬盘
支持TCP/IP的网络硬件设备
软件要求：
操作系统：Linux, Windows 2000 and XP, 等
网络协议：TCP/IP
Java运行环境：JDK1.4
4.3 Joram产品包目录结构
doc/ ——Joram产品帮助文档
samples/ ——Joram产品使用举例
bin/... ——Joram例子执行程序
config/... ——Joram例子配置文件
class/... ——Joram例子class文件
run/... ——Joram例子运行目录
src/ ——Joram产品例子源代码
joram/... ——J2EE环境的Joram源代码
kjoram/… ——J2ME环境的Joram源代码
ship/lib/... ——Joram所用的lib包
ship/licenses/... ——Joram的许可证
5 JORAM Classic Samples使用说明
Joram classic samples使用简单的配置创建一个Message Server，发布一个Queue和一个Topic，并允许匿名用户访问。其中包口：一个消息发送的例子、一个消息发布的例子、一个标准的消息生产者的例子、一个消息接收的例子、一个消息订阅的例子、一个标准的消息消费者的例子、一个浏览Queue的例子。Joram平台运行在稳定模式。
可以利用ant工具直接运行Joram的上述例子
起动Joram消息Server:
ant single_server
也可以用命令行启动Joram Server，进入$JoramHome/samples/bin/目录，运行single_server.bat(win)或single_server.sh(unix)。以下操作均有类似的命令行工具。
建立Queue和Topic:
ant classic_admin
运行Sender发送消息例子:
ant sender
运行Receiver接受消息例子:
ant receiver
运行Browser 浏览Queue例子:
ant browser
运行消息订约的例子:
ant subscriber
运行消息发布的例子:
ant publisher
运行消息消费的例子:
ant consumer
运行消息生产的例子:
ant producer
以上操作也可以利用Eclipse3.1集成开发工具，将Joram例子直接导入成Java工程，然后利用例子中的build.xml文件运行。如图：
右击build.xml文件选择Run As – Ant Build...
弹出窗口如下：每个操作选项后面都有英文的注释信息，说明操作的内容，可以按照前面运行例子的顺序，依次运行即可。
Joram Server也可以通过命令行启动：
进入Joram产品的$JORAM_HOME/samples/bin目录，有下列批处理命令，其中.bat文件是windows环境下的命令，.sh文件是unix环境下的命令。
admin.bat --启动Joram图形控制台界面
admin.sh
classadmin.bat --在启动的Joram Server上创建去queue和topic
clean.bat --清理Joram的运行目录
clean.sh
jmsclient.bat --启动一个JMS客户端
jmsclient.sh
receive.bat --启动一个接收消息进程
send.bat --启动一个发送消息进程
single_server.bat --启动Joram Server
single_server.sh
sslsingle_server.bat
sslsingle_server.sh
server.sh
6 JORAM使用代码举例
6.1 创建消息载体Queue和Topic并绑定Jndi
--用root用户连接Joram Server
AdminModule.connect("root", "root", 60);
--创建Queue和Topic
Queue queue = (Queue) Queue.create("queue");
Topic topic = (Topic) Topic.create("topic");
--创建匿名访问用户
User user = User.create("anonymous", "anonymous");
--设置Queue和Topic为可读、可写
queue.setFreeReading();
topic.setFreeReading();
queue.setFreeWriting();
topic.setFreeWriting();
--创建连接工厂
javax.jms.ConnectionFactory cf =
TcpConnectionFactory.create("localhost", 16010);
javax.jms.QueueConnectionFactory qcf =
QueueTcpConnectionFactory.create("localhost", 16010);
javax.jms.TopicConnectionFactory tcf =
TopicTcpConnectionFactory.create("localhost", 16010);
--邦定Jndi
javax.naming.Context jndiCtx = new javax.naming.InitialContext();
jndiCtx.bind("cf", cf);
jndiCtx.bind("qcf", qcf);
jndiCtx.bind("tcf", tcf);
jndiCtx.bind("queue", queue);
jndiCtx.bind("topic", topic);
jndiCtx.close();
--关闭连接
AdminModule.disconnect();
另一种创建Queue和Topic的方法：
可以将Queue和Topic的具体信息写入一个joramAdmin.XML配置文件，然后在程序中读取这个XML配置文件即可创建。举例如下：
joramAdmin.XML：
==============================================
<?xml version="1.0"?>
<JoramAdmin>
<AdminModule>
<connect host="localhost"
port="16010"
name="root"
password="root"/>
</AdminModule>
<InitialContext>
<property name="java.naming.factory.initial"
value="fr.dyade.aaa.jndi2.client.NamingContextFactory"/>
<property name="java.naming.factory.host" value="localhost"/>
<property name="java.naming.factory.port" value="16400"/>
</InitialContext>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.TcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="cf"/>
</ConnectionFactory>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.TopicTcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="tcf"/>
</ConnectionFactory>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.QueueTcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="qcf"/>
</ConnectionFactory>
<User name="anonymous"
password="anonymous"/>
<Queue name="queue">
<freeReader/>
<freeWriter/>
<jndi name="queue"/>
</Queue>
<Topic name="topic">
<freeReader/>
<freeWriter/>
<jndi name="topic"/>
</Topic>
</JoramAdmin>
==============================================
代码举例：***代表配置文件的具体路径
AdminModule.connect("root", "root", 60);
AdminModule.executeXMLAdmin("***","joramAdmin.xml");
AdminModule.disconnect();
6.2 发送消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--通过Jndi邦定创建好的Queue建立连接工厂
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx
.lookup("qcf");
ictx.close();
--创建Queue连接
QueueConnection qc = qcf.createQueueConnection();
--创建Queue会话
QueueSession qs = qc.createQueueSession(true, 0);
注意：此处创建的Session使用了事务模式，因此在执行具体的发送消息代码即qsend.send()方法后必须要提交Session事务，即：qs.commit()，否则消息不会发送。其他举例代码类似，不再逐一复述。
--创建QueueSender
QueueSender qsend = qs.createSender(queue);
--创建文本消息
TextMessage msg = qs.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
--设置消息具体内容
msg.setText("Test number " + i);
--发送的消息可以设置具体的业务种类，或其他的类型标志：
msg.setIntProperty("ywlx",i);
msg.setIntProperty("****",*)
--发送消息
qsend.send(msg);
}
--提交Queue会话
qs.commit();
System.out.println(i + " messages sent.");
--关闭Queue连接
qc.close();
消息的主体也可以是Java对象：
--创建对象消息
ObjectMessage omsg = qs.createObjectMessage();
omsg.clearBody();
--实例化并传送的Java对象
TestObj to = new TestObj();
omsg.setObject(to);
qsend.send(omsg);
qs.commit();
qc.close();
6.3 接收消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--通过Jndi邦定创建好的Queue建立连接工厂
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx
.lookup("qcf");
ictx.close();
--创建Queue连接
QueueConnection qc = qcf.createQueueConnection();
--创建Queue会话
QueueSession qs = qc.createQueueSession(true, 0);
--创建QueueReceiver实例
QueueReceiver qrec = qs.createReceiver(queue);
--创建消息接收体
Message msg;
TestObj to;
--起动Queue连接
qc.start();
int i;
for (i = 0; i < 10; i++) {
--接收消息
msg = qrec.receive();
--接收文本消息
if (msg instanceof TextMessage)
System.out.println("Msg received: "
+ ((TextMessage) msg).getText());
--接收Java对象消息
else if (msg instanceof ObjectMessage) {
to = (TestObj) ((ObjectMessage) msg).getObject();
to.TestFunc();
} else
System.out.println("Msg received: " + msg);
}
--提交Queue会话
qs.commit();
System.out.println();
System.out.println(i + " messages received.");
--关闭Queue连接
qc.close();
接收指定业务种类的消息：
--指定接收消息的具体类型
此处用到了Message Selector来创建QueueReceiver对象，详细介绍请参看前面的介绍。
String strMsgSelect = "ywlx=0";
QueueConnection qc = (QueueConnection) qcf.createQueueConnection();
QueueSession qs = (QueueSession) qc.createQueueSession(true, 0);
--用MesageSelecter创建QueueReceiver对象
QueueReceiver qrec = (QueueReceiver) qs.createReceiver(queue,
strMsgSelect);
--接收指定的消息；
msg = (Message) qrec.receive();
6.4 发布消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定创建好的Topic建立连接工厂
TopicConnectionFactory tcf = (TopicConnectionFactory) ictx.lookup("tcf");
ictx.close();
--创建Topic连接
TopicConnection tc = tcf.createTopicConnection();
--创建Topic会话
TopicSession ts = tc.createTopicSession(true, 0);
--创建消息发布实例
TopicPublisher tpub = ts.createPublisher(topic);
--创建文本消息
TextMessage msg = ts.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
msg.setText("Test number " + i);
--发布消息
tpub.publish(msg);
}
--提交Topic会话
ts.commit();
System.out.println(i + " messages published.");
--关闭Topic连接
tc.close();
6.5 订阅消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定创建好的Topic建立连接工厂
TopicConnectionFactory tcf = (TopicConnectionFactory) ictx.lookup("tcf");
ictx.close();
--创建Topic连接
TopicConnection tc = tcf.createTopicConnection();
--创建Topic会话
TopicSession ts = tc.createTopicSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
--创建消息订阅实例
TopicSubscriber tsub = ts.createSubscriber(topic);
--订阅消息
tsub.setMessageListener(new MsgListener());
--起动Topic连接
tc.start();
System.in.read();
--关闭Topoc连接
tc.close();
其中MsgListener类是一个实现了MessageListener接口的具体类，它实现了onMessage方法，具体处理监听到的消息。监听启动后，一旦队列中出现了被监听的消息，系统会自动调用MsgListener的onMessage方法去处理消息。
MsgListener类:
public class MsgListener implements MessageListener{
String ident = null;
public MsgListener(){}
public MsgListener(String ident){
this.ident = ident;
}
public void onMessage(Message msg){
try {--根据不同的消息类型接收消息
if (msg instanceof TextMessage) {
if (ident == null)
System.out.println(((TextMessage) msg).getText());
else
System.out.println(ident + ": " + ((TextMessage) msg).getText());
}
else if (msg instanceof ObjectMessage) {
if (ident == null)
System.out.println(((ObjectMessage) msg).getObject());
else
System.out.println(ident + ": " + ((ObjectMessage) msg).getObject());
}
}catch (JMSException jE) {
System.err.println("Exception in listener: " + jE);
}
}
}
6.6 产生消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定建立连接工厂
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
ictx.close();
--创建连接
Connection cnx = cf.createConnection();
--创建会话
Session sess = cnx.createSession(true, 0);
--创建产生消息实例
MessageProducer producer = sess.createProducer(null);
--创建文本消息
TextMessage msg = sess.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
msg.setText("Test number " + i);
--在Queue上产生消息
producer.send(queue, msg);
--在Topic上产生消息
producer.send(topic, msg);
}
--提交会话
sess.commit();
System.out.println(i + " messages sent.");
--关闭连接
cnx.close();
6.7 消费消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定建立连接工厂
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
ictx.close();
--创建连接
Connection cnx = cf.createConnection();
--创建会话
Session sess = cnx.createSession(false, Session.AUTO_ACKNOWLEDGE);
--创建消费消息实例
MessageConsumer recv = sess.createConsumer(queue);
MessageConsumer subs = sess.createConsumer(topic);
--订阅Queue和Topic上的消息
recv.setMessageListener(new MsgListener("Queue listener"));
subs.setMessageListener(new MsgListener("Topic listener"));
--起动连接
cnx.start();
System.in.read();
cnx.close();
System.out.println();
System.out.println("Consumer closed.");
6.8 对Queue的清理操作
--创建连接
AdminModule.connect("root", "root", 60);
--读取server.properties具体配置信息
ictx = new InitialContext(JMSConstant.getProp());
Queue queue = (Queue) ictx.lookup("queue");
--清理Queue
queue.clear();
6.9 Connectiion的注意事项:
在创建jms客户端的时候.有一个问题值得注意.在一个jms运用当中,至少需要两个客户端(消息发送端和消息接手端).每个客户端都需通过jndi去找连接工厂.并且创建一个连接。
消息发送端:
………………………………………………………………………………………….
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection();
……………………………………………………………………………………………...
消息接收端
…………………………………………………………………………………………...
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection();
…………………………………………………………………………………………..
两个客户端都享有一个连接工厂，而且很可能占有同一个连接，如果当接收端需要调用cn.close()关闭连接.cn.start ();打开连接,或者发送段cn.close()关闭连接.cn.start ();就会发生不可预料的情况.
要区别连接是很必要的.所以我们创建2个User.分别如下:
//User user= User.create(String username,PassWord password,Int serverID);
User user1=User.create(“user1”,”user1”,0);
User user2=User.create(“user2”,”user2”,0);
然后根据不同的用户,来创建一个不同的连接.
消息发送端
…………………………………………………………………………………….
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection(“user1”,”user1”);
………………………………………………………………...
消息接收端
………………………………………………………………….
ConnectionFactory cf= (ConnectionFactory) ictx.lookup(“cf”);
Connection cn = cf.createConnection(“user2”,”user2”);
6.10设置消息监听
在消息的接收客户端（包括接收者和订阅者以及消息消费者）都可以设置消息监听,当消息监听监听到消息目的地有消息,就接收消息,注意监听和接收在程序级别上是不可分隔的。JMS提供了一个消息监听接口，只需实现消息监听的onMessage（Message Msg）方法。
代码如下：
创建一个消息监听实现类：
Public class MsgListener implments MessageListener{
Public void onMessage(Message Msg){
…………….//实现代码，处理数据。
}
};
P2p：receiver.setMessageListener(new MsgListener());
p/s: subscriber.setMessageListener(new MsgListener());
注意：一个消息接收端设置一个监听，如果共用一个监听，可能会出现资源同步占用。
onmessage由于
6.11消息接收的过期
在消息接收当中,可以在receiver(Long TimeOut)设置接收消息的过期时间,如果在取消息的时候没有符合过滤器相对应的消息,那么在时间过期后返回一个null值,我们可以对这个null做判断,来关闭连接.这个方法可用于各种消息模式中.
注意:receive()这个动作是一次性的,一个recevie()动作只接收一个消息.消息的单位数也取决与send()这个动作的次数.既是,send()动作一次,那么所发送的消息个数为一.
代码如下:
Session se =new Session();
Receiver re= se.create(Queue queue,String messageselector );
Message msg;
msg = qrec.receive(3000);;//设置消息接收过期时间3秒.
………………………//消息处理
Msg.getText();//如果消息在过期时间内没有接受消息,那么返回一个null值.
如果在接收的时候,不需要等待,那么可以调用recevieNoWait();那么如果在queue没有相应的消息,立即返回一个null.
代码如下:
Session se =new Session();
Receiver re= se.create(Queue queue,String messageselector );
Message msg;
msg = qrec.receiveNoWait();;// 接收消息不等待.
………………………//消息处理
Msg.getText();//如果消息在过期时间内没有接受消息,那么返回一个null值.
6.12.多线程的消息发送
代码说明:首先要在主类里创建一个内部类继承thread的run方法,以实现多线程的发送,在主类里构建一个构造函数,把一些发送消息的预备工作初始化,从而让多线程只处理发送的动作,节省每次线程初始发送的预备动作,来达到快速的发送消息.
这个示例是创建一个消息发送器发送两个分别设置不同属性的消息构造器
public class Sendert {
private static transient final Log log = LogFactory.getLog(Sendert.class);
QueueConnection queueConnection = null;
QueueSession queueSession = null;
QueueSender queueSender = null;
final String MSG_TEXT1 =
new String("Here is a client-acknowledge message 4tetrtretertert4t45ewrwerwrgergrt4rqwerew5t479347597505890 34535&%^*&$&^$^%$&%&*)_*)&^*%*^$&$^(*&(*&(*%*%&^%*%^%*&^*&^&*^*&^*&^");
final String MSG_TEXT2 =
new String("Here is a Message 327893475984 ryeirrfr9urtyureyt&&&&&&&)))__^&&&))gerg)%%^%^%^%&(*(*(*)*(**)(*)*)*)T^HJCJBDKJFBKJJDOAIWEU(Q&#(&E(*&#Q*E&QEUODIJEDLJLHDKGDUWTE&QE");
TextMessage message1 = null;
TextMessage message2 = null;
static long time1;
static long time4;
/**
*构造函数对发送消息以前的动作初始化,从而让这些动作从线程中分离出来,提高代码的效**率.
*/
public Sendert() throws JMSException, NamingException {
//javax.naming.Context ictx = new InitialContext(JMSConstant.getProp());
/**
* 直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码.
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
javax.naming.Context ictx = new InitialContext(pr);
QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) ictx.lookup("qcf");
queueConnection = queueConnectionFactory.createQueueConnection("user1","user1");
queueSession = queueConnection.createQueueSession(false,
Session.CLIENT_ACKNOWLEDGE);
Queue queue = (Queue) ictx.lookup("queue");
ictx.close();
queueSender = queueSession.createSender(queue);//创建消息发送器
message1 = queueSession.createTextMessage();//创建消息生成器
message2 = queueSession.createTextMessage();//创建消息生成器
message1.setIntProperty("sessionID", 1);//设置消息属性.以便消息接收的过滤
message2.setIntProperty("sessionID", 2);//设置消息属性, 以便消息接收的过滤
}
class AsynSender extends Thread {
/**
*继承run方法,发送消息.message1和message2
*/
public void run () {
try {
for(int i=0;i<100;i++){
message1.setText("i"+MSG_TEXT1);//设置消息
message2.setText("i"+MSG_TEXT2);//设置消息
System.out.println(" message1: " + message1.getText());
System.out.println(" message2: " + message2.getText());
queueSender.send(message1);//发送消息
queueSender.send(message2);//发送消息
}
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
}
}
}
/**
* 由于是多线程,无法控制每个线程同时完成,所以会发生没有发送消息完毕而关闭连接的情况
* 省去下面注释的代码.
*/
// finally {
// if(queueSender!=null){
// try{queueSender.close();
//
// }catch(JMSException e){}
// if(queueSession!=null)
// {
// try{
// queueSession.commit();
// }catch(JMSException e){}
// }
// if (queueConnection != null)
// {
// try {
// queueConnection.close();
// } catch (JMSException e) {}
// }
// }
// }
/**
*创建发送消息的线程组.启动线程
*/
public void run_threads() {
AsynSender[] tt = new AsynSender[100];
for (int i = 0; i < tt.length; i++)
{
tt[i] = new AsynSender();
tt[i].start();
}
for (int i = 0; i < tt.length; i++) {
System.out.println("go1");
try {
tt[i].join();//让子线程发送消息完毕后,才执行主控线程,关闭发送器,连接,会话对象
} catch (InterruptedException je)
{
Je.printstacktrace();}
}
}
/**
* 创建方法close()关闭消息发送器,会话,连接对象.
*
* @throws JMSException
*/
void close() throws JMSException {
queueSender.close();
queueSession.close();
queueConnection.close();
}
/**
*
* @param args
* @throws JMSException
* @throws NamingException
* @throws IOException
* 主控线程,当子线程发送消息完毕后.调用close()方法关闭消息发送器,会话,连接对象.
* 最后打印出发送消息的时间.和一些活动的线程测试.
*
*/
public static void main(String[] args) throws JMSException, NamingException, IOException {
System.out.println(" sender: Created client-acknowledge session");
Sendert se = new Sendert();
se.run_threads();
se.close();//发送消息完毕关闭消息发送器,会话,连接对象.
}
}
6.13 joram远程部署.
Joarm的远程部署,即jms服务器和jms客户端不在同一ip主机上,首先我们选定一服务器主机名是KBF_BCR, ip地址是82.0.176.214.在这台主机上安装joram,配置好服务器的a3server.xml.文件如下:
<?xml version="1.0"?>
<config>
<property name="Transaction" value="fr.dyade.aaa.util.NullTransaction"/>
<server id="0" name="S0" hostname="KBF_BCR">
<service class="org.objectweb.joram.mom.proxies.ConnectionManager"
args="root root"/>
<service class="org.objectweb.joram.mom.proxies.tcp.TcpProxyService"
args="16010"/>
<service class="fr.dyade.aaa.jndi2.server.JndiServer" args="16400"/>
</server>
</config>
启动服务器.就ok了!
二:接下来是在选定另一台主机做为jms的客户端:我们执行如下代码,在服务器下创建queue,topic,user,和jndi对它们的绑顶。代码如下:
public class Amdinx {
public static void main(String args[])throws Exception{
System.out.println("begin the admin test!");
//connecting to JORAM server
AdminModule.connect("82.0.176.214",16010,"root", "root", 60);
// AdminModule连接服务器,参数为服务器ip地址,连接工厂的端口,登陆用户名,密码和每
//秒尝试连接的次数.
Queue queue = (Queue) Queue.create(0);//参数0为服务器id,与a3server.xml对应.
User user1 = User.create("user1","user1",0); //参数0为服务器id,与a3server.xml对应.
User user2 = User.create("user2","user2",0); //参数0为服务器id,与a3server.xml对应.
User user3 = User.create("user3","user3",0); //参数0为服务器id,与a3server.xml对应.
queue.setFreeReading(user1);//只让user1读
queue.setFreeWriting(user1);//只让user1写
javax.jms.QueueConnectionFactory qcf =
QueueTcpConnectionFactory.create("82.0.176.214", 16010);//服务器的连接工厂,参数为服务器ip,和连接工厂的端口号.
/**
*直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
javax.naming.Context jndiCtx = new InitialContext(pr);
// javax.naming.Context jndiCtx = new javax.naming.InitialContext(JMSConstant.getProp());
/**
*绑定对象.
*/
jndiCtx.bind("qcf", qcf);
jndiCtx.bind("queue", queue);
jndiCtx.close();//关闭上下文目录
AdminModule.disconnect();//关闭AdminModule连接
System.out.println("close the admin!");
}
}
这样就执行代码后,就在服务器上创建了用户,目的地,和对它们的jndi绑定.
三:然后执行多线程的发送消息的代码:(见6.12多线程的消息发送)
四:多线程的接收消息,代码如下.
public class Receiver1
{
QueueReceiver qrec1=null;
QueueSession qs =null;
QueueConnection qc=null;
/**
*
* @throws JMSException
* 构造函数初始化
*/
public Receiver1() throws NamingException, JMSException{
//ictx = new InitialContext(JMSConstant.getProp());
/**
*直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
Context ictx = new InitialContext(pr);
Queue queue = (Queue) ictx.lookup("queue");
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx.lookup("qcf");
ictx.close();
qc = qcf.createQueueConnection("user1","user1");//创建只属于user1的连接,以区别其他连接.
qs = qc.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
qrec1 = qs.createReceiver(queue,"sessionID=2");
}
/**
*
* @author Owner
* 一个内部类继承Thread实现run()方法,在run()方法设置消息监听器.
*/
class Brun extends Thread{
public void run(){
try{
qrec1.setMessageListener(new MsgListener1());
qc.start();//把连接开始应该放在此处.设置消息监听动作之后
}catch(JMSException e){}
}
}
/**
* 创建线程组,启动线程组.
*
*/
public void run_thread() {
Brun[] tt = new Brun[300];
for (int i = 0; i < tt.length; i++) {
tt[i] = new Brun();
tt[i].start();
}
}
/**
* 创建方法close()关闭消息发送器,会话,连接对象.
*
* @throws JMSException
*/
void close() throws JMSException {
qrec1.close();
qs.close();
qc.close();
}
/**
*
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception
{
System.out.println("Requests to receive messages...");
Receiver1 rc=new Receiver1();
rc.run_thread();
// qc.start(); 不能把连接开始放在此处,而是放在设置消息监听的动作之后.
System.in .read();//当消息接收完毕,在控制台上输入任意字符,执行关闭的动作.
rc.close();
System.out.println( " messages received.");
}
}
7 joram和spring的集成
7.1 简单的消息发送
一.首先启动joram服务器,管理各种对象和jndi绑定(详细请看前面章节).
二.在工程目录下引入spring支持的包和joram包.
三.在工程目录下创建applicationContext.xml文件.这个文件是spring的配置文件.示例如下
<!-----xml文件头,为默认,我们不需要修改---------------->
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
<!------配置文件内容------->
<beans>
<!------连接工厂配置.注意:value值为jndi绑定后的key=”qcf”------->
<bean id="connectionFactory"
class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>qcf</value>
</property>
<property name="jndiTemplate">
<ref local="jndiTemplate"></ref>
</property>
</bean>
<!-----目的地配置,注意:queue为jndi绑定后的key而不是value>
<bean id="destination"
class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>queue</value>
</property>
<property name="jndiTemplate">
<ref local="jndiTemplate"></ref>
</property>
</bean>
<!---spring-jms模版:注意:joram是支持jms1.1的所以用org.springframework.jms.core.JmsTemplate而不是用org.springframework.jms.core.JmsTemplate102------- >
<bean id="jmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<ref local="connectionFactory" />
</property>
<property name="defaultDestination">
<ref local="destination" />
</property>
</bean>
<!-----配置jndi环境------->
<bean id="jndiTemplate"
class="org.springframework.jndi.JndiTemplate">
<property name="environment">
<props>
<prop key="java.naming.factory.initial">
fr.dyade.aaa.jndi2.client.NamingContextFactory
</prop>
<prop key="java.naming.factory.host">localhost</prop>
<prop key="java.naming.factory.port">16400</prop>
</props>
</property>
</bean>
</beans>
四.发送消息客户端代码:
public class JmsTemplateTest11 {
protected static final Log log = LogFactory.getLog(JmsTemplateTest11.class);
static Context ictx = null;
public static void main(String[] args) throws NamingException, JMSException {
System.out.println();
System.out.println("Sends messages on the queue...");
/**引入配置文件,推荐使用FileSystemXmlApplicationContext,这样你可以一次使用
*多个配置文件
*/.
ApplicationContext ac = new FileSystemXmlApplicationContext(
new String[]{
"F:\\project\\SpringJMS\\src\\applicationContext.xml"});
JmsTemplate jt = (JmsTemplate) ac.getBean("jmsTemplate");
//注入bean,由于配置文件各个bean是相互关联的,所以注入jmsTemplate就注入了//其它bean,当然你也可以这样写.但是配置文件中各个bean没有关联.这段代码可以//更深入理解spring的ioc模式
//Destination de = (Destination) ac.getBean(“destination”)
//QueueConnetionfactory cf =( QueueConnectionFactory) //ac.getBean(“connectionFactory”)
/**通过jt模版发送消息,这里new MessageCreator()使用的反射原理.
*/
jt.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
// return session
// .createTextMessage("Hello World(11),queue!");
TextMessage message = (TextMessage) session.createTextMessage("bbc");//创建一个消息,内容为bbc.
return message;
}
});
log.info("成功发送消息！");
System.out.println("Success Sends messages to the queue...");
}
}
1.1 编写目的
本文作为B2bi项目中开源产品JORAM的使用指导文档，旨在帮助项目组人员方便明了的进行JMS模块的详细设计和开发工作。本文档主要包含建设银行EAI平台B2Bi子系统中使用的开源JMS产品——JORAM的使用说明。
1.2 名词解释
B2Bi：
Business to Business integration （企业间集成）
JMS:
Java Message Service （Java消息服务）
JORAM:
ObjectWeb的Java开源项目
JNDI：
Java命名和目录接口
1.3 参考资料
《Joram-4.3-en.pdf》——JORAM使用手册（英文）
《Joram4_0_SAMPLES.pdf》——JORAM使用举例（英文）
《Joram4_0_ADMIN.pdf》——JORAM管理员手册（英文）
2 JMS简介
2.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。
2.2 JMS基本功能
JMS是用于和面向消息的中间件相互通信的应用程序接口。它既支持点对点(point-to-point)的域，又支持发布/订阅(publish/subscribe)类型的域，并且提供对下列类型的支持：经认可的消息传递,事务型消息的传递，一致性消息和具有持久性的订阅者支持。JMS还提供了另一种方式来对您的应用与旧的后台系统相集成。
2.3 消息服务类型
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相关文档。
2.4 JMS接口类图
2.5 JMS基本视图
1.JMS接口描述
JMS 支持两种消息类型PTP 和Pub/Sub，分别称作：PTP Domain 和Pub/Sub Domain，这两种接口都继承统一的JMS父接口，JMS 主要接口如下所示：
MS父接口 PTP Pub/Sub
ConnectionFactory QueueConnectionFactory TopicConnectionFactory
Connection QueueConnection TopicConnection
Destination Queue Topic
Session QueueSession TopicSession
MessageProducer QueueSender TopicPublisher
MessageConsumer QueueReceiver,QueueBrowse r TopicSubscriber
ConnectionFactory ：连接工厂，JMS 用它创建连接
Connection ：JMS 客户端到JMS Provider 的连接
Destination ：消息的目的地
Session： 一个发送或接收消息的线程
MessageProducer： 由Session 对象创建的用来发送消息的对象
MessageConsumer： 由Session 对象创建的用来接收消息的对象
2.JMS消息模型
JMS 消息由以下几部分组成：消息头，属性，消息体。
2.1 消息头(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
2.2 属性(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 严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。
2.3 消息体(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
}
3. 消息的同步异步接收
消息的同步接收是指客户端主动去接收消息，JMS 客户端可以采用MessageConsumer 的receive方法去接收下一个消息。
消息的异步接收是指当消息到达时，主动通知客户端。JMS 客户端可以通过注册一个实 现MessageListener 接口的对象到MessageConsumer，这样，每当消息到达时，JMS Provider 会调用MessageListener中的onMessage 方法。
4. PTP模型
PTP(Point-to-Point)模型是基于队列的，发送方发消息到队列，接收方从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提 供工具管理队列的创建、删除。JMS 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 浏览队列中的消息，但不会收走消息。
QueueRequestor JMS 提供QueueRequestor 类简化消息的收发过程。QueueRequestor 的构造函数有两个参数：QueueSession 和queue，QueueRequestor 通过创建一个临时队列来完成最终的收发消息请求。
可靠性(Reliability) 队列可以长久地保存消息直到接收者收到消息。接收者不需要因为担心消息会丢失而时刻和队列保持激活的连接状态，充分体现了异步传输模式的优势。
5. PUB/SUB模型
JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作主题(topic)。
主题可以被认为是消息的传输中介，发布者(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 对象。
TopicConnection TopicConnection 是一个到JMS Pub/Sub provider 的连接，客户端可以用TopicConnection创建TopicSession 来发布和订阅消息。
TopicSession TopicSession 提供一些方法创建TopicPublisher、TopicSubscriber、TemporaryTopic 。它还提供unsubscribe 方法取消消息的持久订阅。
TopicPublisher 客户端用TopicPublisher 发布消息到主题。
TopicSubscriber 客户端用TopicSubscriber 接收发布到主题上的消息。可以在TopicSubscriber 中设置消息过滤功能，这样，不符合要求的消息不会被接收。
Durable TopicSubscriber 如果一个客户端需要持久订阅消息，可以使用Durable TopicSubscriber，TopSession 提供一个方法createDurableSubscriber创建Durable TopicSubscriber 对象。
恢复和重新派送(Recovery and Redelivery) 非持久订阅状态下，不能恢复或重新派送一个未签收的消息。只有持久订阅才能恢复或重新派送一个未签收的消息。
TopicRequestor JMS 提供TopicRequestor 类简化消息的收发过程。TopicRequestor 的构造函数有两个参数：TopicSession 和topic。TopicRequestor 通过创建一个临时主题来完成最终的发布和接收消息请求。
可靠性(Reliability) 当所有的消息必须被接收，则用持久订阅模式。当丢失消息能够被容忍，则用非持久订阅模式。
3 JMS API编程模型
一个JMS应用由以下几个模块组成：
3.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");
3.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接口的实例。
3.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后不自动确认。
3.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);
3.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。
a) 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在运行。
b) 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的内容进行选择。
3.6 Message—消息组成
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 没有消息体，只有消息头和属性。
JMS API为每一种messages都提供了一种create方法。例如：
TextMessage message = queueSession.createTextMessage();
message.setText(msg_text); // msg_text is a String
queueSender.send(message);
在使用者一端, 必须将收到的Message 按照适当的message类型处理。 例如:
Message m = queueReceiver.receive();
if (m instanceof TextMessage) {
TextMessage message = (TextMessage) m;
System.out.println("Reading message: " + message.getText());
} else {
// Handle error
}
3.7 Exception Handling
JMS API的方法的Exception的根类是JMSException。JMSException类包括如下子类:
IllegalStateException
InvalidClientIDException
InvalidDestinationException
InvalidSelectorException
JMSSecurityException
MessageEOFException
MessageFormatException
MessageNotReadableException
MessageNotWriteableException
ResourceAllocationException
TransactionInProgressException
TransactionRolledBackException
4 JORAM产品介绍
JORAM是法国电信研究院支持的objectweb.org的开源JMS消息中间件产品。objectweb是一个非常活跃的开源中间件团体。JORAM支持sun规范，实现JMS1.1。
4.1 Joram4.3包括
一个Message Server（消息服务器），提供Java的消息服务功能。
一个JNDI Server。
一个客户端类，可以访问Message Server上的消息。
一个图形化的管理界面，配置管理Joram平台。
4.2 系统要求
硬件要求：
2000年后的intel架构pc机
256M RAM 5G硬盘
支持TCP/IP的网络硬件设备
软件要求：
操作系统：Linux, Windows 2000 and XP, 等
网络协议：TCP/IP
Java运行环境：JDK1.4
4.3 Joram产品包目录结构
doc/ ——Joram产品帮助文档
samples/ ——Joram产品使用举例
bin/... ——Joram例子执行程序
config/... ——Joram例子配置文件
class/... ——Joram例子class文件
run/... ——Joram例子运行目录
src/ ——Joram产品例子源代码
joram/... ——J2EE环境的Joram源代码
kjoram/… ——J2ME环境的Joram源代码
ship/lib/... ——Joram所用的lib包
ship/licenses/... ——Joram的许可证
5 JORAM Classic Samples使用说明
Joram classic samples使用简单的配置创建一个Message Server，发布一个Queue和一个Topic，并允许匿名用户访问。其中包口：一个消息发送的例子、一个消息发布的例子、一个标准的消息生产者的例子、一个消息接收的例子、一个消息订阅的例子、一个标准的消息消费者的例子、一个浏览Queue的例子。Joram平台运行在稳定模式。
可以利用ant工具直接运行Joram的上述例子
起动Joram消息Server:
ant single_server
也可以用命令行启动Joram Server，进入$JoramHome/samples/bin/目录，运行single_server.bat(win)或single_server.sh(unix)。以下操作均有类似的命令行工具。
建立Queue和Topic:
ant classic_admin
运行Sender发送消息例子:
ant sender
运行Receiver接受消息例子:
ant receiver
运行Browser 浏览Queue例子:
ant browser
运行消息订约的例子:
ant subscriber
运行消息发布的例子:
ant publisher
运行消息消费的例子:
ant consumer
运行消息生产的例子:
ant producer
以上操作也可以利用Eclipse3.1集成开发工具，将Joram例子直接导入成Java工程，然后利用例子中的build.xml文件运行。如图：
右击build.xml文件选择Run As – Ant Build...
弹出窗口如下：每个操作选项后面都有英文的注释信息，说明操作的内容，可以按照前面运行例子的顺序，依次运行即可。
Joram Server也可以通过命令行启动：
进入Joram产品的$JORAM_HOME/samples/bin目录，有下列批处理命令，其中.bat文件是windows环境下的命令，.sh文件是unix环境下的命令。
admin.bat --启动Joram图形控制台界面
admin.sh
classadmin.bat --在启动的Joram Server上创建去queue和topic
clean.bat --清理Joram的运行目录
clean.sh
jmsclient.bat --启动一个JMS客户端
jmsclient.sh
receive.bat --启动一个接收消息进程
send.bat --启动一个发送消息进程
single_server.bat --启动Joram Server
single_server.sh
sslsingle_server.bat
sslsingle_server.sh
server.sh
6 JORAM使用代码举例
6.1 创建消息载体Queue和Topic并绑定Jndi
--用root用户连接Joram Server
AdminModule.connect("root", "root", 60);
--创建Queue和Topic
Queue queue = (Queue) Queue.create("queue");
Topic topic = (Topic) Topic.create("topic");
--创建匿名访问用户
User user = User.create("anonymous", "anonymous");
--设置Queue和Topic为可读、可写
queue.setFreeReading();
topic.setFreeReading();
queue.setFreeWriting();
topic.setFreeWriting();
--创建连接工厂
javax.jms.ConnectionFactory cf =
TcpConnectionFactory.create("localhost", 16010);
javax.jms.QueueConnectionFactory qcf =
QueueTcpConnectionFactory.create("localhost", 16010);
javax.jms.TopicConnectionFactory tcf =
TopicTcpConnectionFactory.create("localhost", 16010);
--邦定Jndi
javax.naming.Context jndiCtx = new javax.naming.InitialContext();
jndiCtx.bind("cf", cf);
jndiCtx.bind("qcf", qcf);
jndiCtx.bind("tcf", tcf);
jndiCtx.bind("queue", queue);
jndiCtx.bind("topic", topic);
jndiCtx.close();
--关闭连接
AdminModule.disconnect();
另一种创建Queue和Topic的方法：
可以将Queue和Topic的具体信息写入一个joramAdmin.XML配置文件，然后在程序中读取这个XML配置文件即可创建。举例如下：
joramAdmin.XML：
==============================================
<?xml version="1.0"?>
<JoramAdmin>
<AdminModule>
<connect host="localhost"
port="16010"
name="root"
password="root"/>
</AdminModule>
<InitialContext>
<property name="java.naming.factory.initial"
value="fr.dyade.aaa.jndi2.client.NamingContextFactory"/>
<property name="java.naming.factory.host" value="localhost"/>
<property name="java.naming.factory.port" value="16400"/>
</InitialContext>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.TcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="cf"/>
</ConnectionFactory>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.TopicTcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="tcf"/>
</ConnectionFactory>
<ConnectionFactory className="org.objectweb.joram.client.jms.tcp.QueueTcpConnectionFactory">
<tcp host="localhost"
port="16010"/>
<jndi name="qcf"/>
</ConnectionFactory>
<User name="anonymous"
password="anonymous"/>
<Queue name="queue">
<freeReader/>
<freeWriter/>
<jndi name="queue"/>
</Queue>
<Topic name="topic">
<freeReader/>
<freeWriter/>
<jndi name="topic"/>
</Topic>
</JoramAdmin>
==============================================
代码举例：***代表配置文件的具体路径
AdminModule.connect("root", "root", 60);
AdminModule.executeXMLAdmin("***","joramAdmin.xml");
AdminModule.disconnect();
6.2 发送消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--通过Jndi邦定创建好的Queue建立连接工厂
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx
.lookup("qcf");
ictx.close();
--创建Queue连接
QueueConnection qc = qcf.createQueueConnection();
--创建Queue会话
QueueSession qs = qc.createQueueSession(true, 0);
注意：此处创建的Session使用了事务模式，因此在执行具体的发送消息代码即qsend.send()方法后必须要提交Session事务，即：qs.commit()，否则消息不会发送。其他举例代码类似，不再逐一复述。
--创建QueueSender
QueueSender qsend = qs.createSender(queue);
--创建文本消息
TextMessage msg = qs.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
--设置消息具体内容
msg.setText("Test number " + i);
--发送的消息可以设置具体的业务种类，或其他的类型标志：
msg.setIntProperty("ywlx",i);
msg.setIntProperty("****",*)
--发送消息
qsend.send(msg);
}
--提交Queue会话
qs.commit();
System.out.println(i + " messages sent.");
--关闭Queue连接
qc.close();
消息的主体也可以是Java对象：
--创建对象消息
ObjectMessage omsg = qs.createObjectMessage();
omsg.clearBody();
--实例化并传送的Java对象
TestObj to = new TestObj();
omsg.setObject(to);
qsend.send(omsg);
qs.commit();
qc.close();
6.3 接收消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--通过Jndi邦定创建好的Queue建立连接工厂
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx
.lookup("qcf");
ictx.close();
--创建Queue连接
QueueConnection qc = qcf.createQueueConnection();
--创建Queue会话
QueueSession qs = qc.createQueueSession(true, 0);
--创建QueueReceiver实例
QueueReceiver qrec = qs.createReceiver(queue);
--创建消息接收体
Message msg;
TestObj to;
--起动Queue连接
qc.start();
int i;
for (i = 0; i < 10; i++) {
--接收消息
msg = qrec.receive();
--接收文本消息
if (msg instanceof TextMessage)
System.out.println("Msg received: "
+ ((TextMessage) msg).getText());
--接收Java对象消息
else if (msg instanceof ObjectMessage) {
to = (TestObj) ((ObjectMessage) msg).getObject();
to.TestFunc();
} else
System.out.println("Msg received: " + msg);
}
--提交Queue会话
qs.commit();
System.out.println();
System.out.println(i + " messages received.");
--关闭Queue连接
qc.close();
接收指定业务种类的消息：
--指定接收消息的具体类型
此处用到了Message Selector来创建QueueReceiver对象，详细介绍请参看前面的介绍。
String strMsgSelect = "ywlx=0";
QueueConnection qc = (QueueConnection) qcf.createQueueConnection();
QueueSession qs = (QueueSession) qc.createQueueSession(true, 0);
--用MesageSelecter创建QueueReceiver对象
QueueReceiver qrec = (QueueReceiver) qs.createReceiver(queue,
strMsgSelect);
--接收指定的消息；
msg = (Message) qrec.receive();
6.4 发布消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定创建好的Topic建立连接工厂
TopicConnectionFactory tcf = (TopicConnectionFactory) ictx.lookup("tcf");
ictx.close();
--创建Topic连接
TopicConnection tc = tcf.createTopicConnection();
--创建Topic会话
TopicSession ts = tc.createTopicSession(true, 0);
--创建消息发布实例
TopicPublisher tpub = ts.createPublisher(topic);
--创建文本消息
TextMessage msg = ts.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
msg.setText("Test number " + i);
--发布消息
tpub.publish(msg);
}
--提交Topic会话
ts.commit();
System.out.println(i + " messages published.");
--关闭Topic连接
tc.close();
6.5 订阅消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定创建好的Topic建立连接工厂
TopicConnectionFactory tcf = (TopicConnectionFactory) ictx.lookup("tcf");
ictx.close();
--创建Topic连接
TopicConnection tc = tcf.createTopicConnection();
--创建Topic会话
TopicSession ts = tc.createTopicSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
--创建消息订阅实例
TopicSubscriber tsub = ts.createSubscriber(topic);
--订阅消息
tsub.setMessageListener(new MsgListener());
--起动Topic连接
tc.start();
System.in.read();
--关闭Topoc连接
tc.close();
其中MsgListener类是一个实现了MessageListener接口的具体类，它实现了onMessage方法，具体处理监听到的消息。监听启动后，一旦队列中出现了被监听的消息，系统会自动调用MsgListener的onMessage方法去处理消息。
MsgListener类:
public class MsgListener implements MessageListener{
String ident = null;
public MsgListener(){}
public MsgListener(String ident){
this.ident = ident;
}
public void onMessage(Message msg){
try {--根据不同的消息类型接收消息
if (msg instanceof TextMessage) {
if (ident == null)
System.out.println(((TextMessage) msg).getText());
else
System.out.println(ident + ": " + ((TextMessage) msg).getText());
}
else if (msg instanceof ObjectMessage) {
if (ident == null)
System.out.println(((ObjectMessage) msg).getObject());
else
System.out.println(ident + ": " + ((ObjectMessage) msg).getObject());
}
}catch (JMSException jE) {
System.err.println("Exception in listener: " + jE);
}
}
}
6.6 产生消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定建立连接工厂
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
ictx.close();
--创建连接
Connection cnx = cf.createConnection();
--创建会话
Session sess = cnx.createSession(true, 0);
--创建产生消息实例
MessageProducer producer = sess.createProducer(null);
--创建文本消息
TextMessage msg = sess.createTextMessage();
int i;
for (i = 0; i < 10; i++) {
msg.setText("Test number " + i);
--在Queue上产生消息
producer.send(queue, msg);
--在Topic上产生消息
producer.send(topic, msg);
}
--提交会话
sess.commit();
System.out.println(i + " messages sent.");
--关闭连接
cnx.close();
6.7 消费消息
--初始化上下文空间
ictx = new InitialContext();
--实例化Queue对象
Queue queue = (Queue) ictx.lookup("queue");
--实例化Topic对象
Topic topic = (Topic) ictx.lookup("topic");
--通过Jndi邦定建立连接工厂
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
ictx.close();
--创建连接
Connection cnx = cf.createConnection();
--创建会话
Session sess = cnx.createSession(false, Session.AUTO_ACKNOWLEDGE);
--创建消费消息实例
MessageConsumer recv = sess.createConsumer(queue);
MessageConsumer subs = sess.createConsumer(topic);
--订阅Queue和Topic上的消息
recv.setMessageListener(new MsgListener("Queue listener"));
subs.setMessageListener(new MsgListener("Topic listener"));
--起动连接
cnx.start();
System.in.read();
cnx.close();
System.out.println();
System.out.println("Consumer closed.");
6.8 对Queue的清理操作
--创建连接
AdminModule.connect("root", "root", 60);
--读取server.properties具体配置信息
ictx = new InitialContext(JMSConstant.getProp());
Queue queue = (Queue) ictx.lookup("queue");
--清理Queue
queue.clear();
6.9 Connectiion的注意事项:
在创建jms客户端的时候.有一个问题值得注意.在一个jms运用当中,至少需要两个客户端(消息发送端和消息接手端).每个客户端都需通过jndi去找连接工厂.并且创建一个连接。
消息发送端:
………………………………………………………………………………………….
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection();
……………………………………………………………………………………………...
消息接收端
…………………………………………………………………………………………...
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection();
…………………………………………………………………………………………..
两个客户端都享有一个连接工厂，而且很可能占有同一个连接，如果当接收端需要调用cn.close()关闭连接.cn.start ();打开连接,或者发送段cn.close()关闭连接.cn.start ();就会发生不可预料的情况.
要区别连接是很必要的.所以我们创建2个User.分别如下:
//User user= User.create(String username,PassWord password,Int serverID);
User user1=User.create(“user1”,”user1”,0);
User user2=User.create(“user2”,”user2”,0);
然后根据不同的用户,来创建一个不同的连接.
消息发送端
…………………………………………………………………………………….
ConnectionFactory cf = (ConnectionFactory) ictx.lookup("cf");
Connection cn = cf.createConnection(“user1”,”user1”);
………………………………………………………………...
消息接收端
………………………………………………………………….
ConnectionFactory cf= (ConnectionFactory) ictx.lookup(“cf”);
Connection cn = cf.createConnection(“user2”,”user2”);
6.10设置消息监听
在消息的接收客户端（包括接收者和订阅者以及消息消费者）都可以设置消息监听,当消息监听监听到消息目的地有消息,就接收消息,注意监听和接收在程序级别上是不可分隔的。JMS提供了一个消息监听接口，只需实现消息监听的onMessage（Message Msg）方法。
代码如下：
创建一个消息监听实现类：
Public class MsgListener implments MessageListener{
Public void onMessage(Message Msg){
…………….//实现代码，处理数据。
}
};
P2p：receiver.setMessageListener(new MsgListener());
p/s: subscriber.setMessageListener(new MsgListener());
注意：一个消息接收端设置一个监听，如果共用一个监听，可能会出现资源同步占用。
onmessage由于
6.11消息接收的过期
在消息接收当中,可以在receiver(Long TimeOut)设置接收消息的过期时间,如果在取消息的时候没有符合过滤器相对应的消息,那么在时间过期后返回一个null值,我们可以对这个null做判断,来关闭连接.这个方法可用于各种消息模式中.
注意:receive()这个动作是一次性的,一个recevie()动作只接收一个消息.消息的单位数也取决与send()这个动作的次数.既是,send()动作一次,那么所发送的消息个数为一.
代码如下:
Session se =new Session();
Receiver re= se.create(Queue queue,String messageselector );
Message msg;
msg = qrec.receive(3000);;//设置消息接收过期时间3秒.
………………………//消息处理
Msg.getText();//如果消息在过期时间内没有接受消息,那么返回一个null值.
如果在接收的时候,不需要等待,那么可以调用recevieNoWait();那么如果在queue没有相应的消息,立即返回一个null.
代码如下:
Session se =new Session();
Receiver re= se.create(Queue queue,String messageselector );
Message msg;
msg = qrec.receiveNoWait();;// 接收消息不等待.
………………………//消息处理
Msg.getText();//如果消息在过期时间内没有接受消息,那么返回一个null值.
6.12.多线程的消息发送
代码说明:首先要在主类里创建一个内部类继承thread的run方法,以实现多线程的发送,在主类里构建一个构造函数,把一些发送消息的预备工作初始化,从而让多线程只处理发送的动作,节省每次线程初始发送的预备动作,来达到快速的发送消息.
这个示例是创建一个消息发送器发送两个分别设置不同属性的消息构造器
public class Sendert {
private static transient final Log log = LogFactory.getLog(Sendert.class);
QueueConnection queueConnection = null;
QueueSession queueSession = null;
QueueSender queueSender = null;
final String MSG_TEXT1 =
new String("Here is a client-acknowledge message 4tetrtretertert4t45ewrwerwrgergrt4rqwerew5t479347597505890 34535&%^*&$&^$^%$&%&*)_*)&^*%*^$&$^(*&(*&(*%*%&^%*%^%*&^*&^&*^*&^*&^");
final String MSG_TEXT2 =
new String("Here is a Message 327893475984 ryeirrfr9urtyureyt&&&&&&&)))__^&&&))gerg)%%^%^%^%&(*(*(*)*(**)(*)*)*)T^HJCJBDKJFBKJJDOAIWEU(Q&#(&E(*&#Q*E&QEUODIJEDLJLHDKGDUWTE&QE");
TextMessage message1 = null;
TextMessage message2 = null;
static long time1;
static long time4;
/**
*构造函数对发送消息以前的动作初始化,从而让这些动作从线程中分离出来,提高代码的效**率.
*/
public Sendert() throws JMSException, NamingException {
//javax.naming.Context ictx = new InitialContext(JMSConstant.getProp());
/**
* 直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码.
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
javax.naming.Context ictx = new InitialContext(pr);
QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) ictx.lookup("qcf");
queueConnection = queueConnectionFactory.createQueueConnection("user1","user1");
queueSession = queueConnection.createQueueSession(false,
Session.CLIENT_ACKNOWLEDGE);
Queue queue = (Queue) ictx.lookup("queue");
ictx.close();
queueSender = queueSession.createSender(queue);//创建消息发送器
message1 = queueSession.createTextMessage();//创建消息生成器
message2 = queueSession.createTextMessage();//创建消息生成器
message1.setIntProperty("sessionID", 1);//设置消息属性.以便消息接收的过滤
message2.setIntProperty("sessionID", 2);//设置消息属性, 以便消息接收的过滤
}
class AsynSender extends Thread {
/**
*继承run方法,发送消息.message1和message2
*/
public void run () {
try {
for(int i=0;i<100;i++){
message1.setText("i"+MSG_TEXT1);//设置消息
message2.setText("i"+MSG_TEXT2);//设置消息
System.out.println(" message1: " + message1.getText());
System.out.println(" message2: " + message2.getText());
queueSender.send(message1);//发送消息
queueSender.send(message2);//发送消息
}
} catch (JMSException e) {
System.out.println("Exception occurred: " + e.toString());
}
}
}
/**
* 由于是多线程,无法控制每个线程同时完成,所以会发生没有发送消息完毕而关闭连接的情况
* 省去下面注释的代码.
*/
// finally {
// if(queueSender!=null){
// try{queueSender.close();
//
// }catch(JMSException e){}
// if(queueSession!=null)
// {
// try{
// queueSession.commit();
// }catch(JMSException e){}
// }
// if (queueConnection != null)
// {
// try {
// queueConnection.close();
// } catch (JMSException e) {}
// }
// }
// }
/**
*创建发送消息的线程组.启动线程
*/
public void run_threads() {
AsynSender[] tt = new AsynSender[100];
for (int i = 0; i < tt.length; i++)
{
tt[i] = new AsynSender();
tt[i].start();
}
for (int i = 0; i < tt.length; i++) {
System.out.println("go1");
try {
tt[i].join();//让子线程发送消息完毕后,才执行主控线程,关闭发送器,连接,会话对象
} catch (InterruptedException je)
{
Je.printstacktrace();}
}
}
/**
* 创建方法close()关闭消息发送器,会话,连接对象.
*
* @throws JMSException
*/
void close() throws JMSException {
queueSender.close();
queueSession.close();
queueConnection.close();
}
/**
*
* @param args
* @throws JMSException
* @throws NamingException
* @throws IOException
* 主控线程,当子线程发送消息完毕后.调用close()方法关闭消息发送器,会话,连接对象.
* 最后打印出发送消息的时间.和一些活动的线程测试.
*
*/
public static void main(String[] args) throws JMSException, NamingException, IOException {
System.out.println(" sender: Created client-acknowledge session");
Sendert se = new Sendert();
se.run_threads();
se.close();//发送消息完毕关闭消息发送器,会话,连接对象.
}
}
6.13 joram远程部署.
Joarm的远程部署,即jms服务器和jms客户端不在同一ip主机上,首先我们选定一服务器主机名是KBF_BCR, ip地址是82.0.176.214.在这台主机上安装joram,配置好服务器的a3server.xml.文件如下:
<?xml version="1.0"?>
<config>
<property name="Transaction" value="fr.dyade.aaa.util.NullTransaction"/>
<server id="0" name="S0" hostname="KBF_BCR">
<service class="org.objectweb.joram.mom.proxies.ConnectionManager"
args="root root"/>
<service class="org.objectweb.joram.mom.proxies.tcp.TcpProxyService"
args="16010"/>
<service class="fr.dyade.aaa.jndi2.server.JndiServer" args="16400"/>
</server>
</config>
启动服务器.就ok了!
二:接下来是在选定另一台主机做为jms的客户端:我们执行如下代码,在服务器下创建queue,topic,user,和jndi对它们的绑顶。代码如下:
public class Amdinx {
public static void main(String args[])throws Exception{
System.out.println("begin the admin test!");
//connecting to JORAM server
AdminModule.connect("82.0.176.214",16010,"root", "root", 60);
// AdminModule连接服务器,参数为服务器ip地址,连接工厂的端口,登陆用户名,密码和每
//秒尝试连接的次数.
Queue queue = (Queue) Queue.create(0);//参数0为服务器id,与a3server.xml对应.
User user1 = User.create("user1","user1",0); //参数0为服务器id,与a3server.xml对应.
User user2 = User.create("user2","user2",0); //参数0为服务器id,与a3server.xml对应.
User user3 = User.create("user3","user3",0); //参数0为服务器id,与a3server.xml对应.
queue.setFreeReading(user1);//只让user1读
queue.setFreeWriting(user1);//只让user1写
javax.jms.QueueConnectionFactory qcf =
QueueTcpConnectionFactory.create("82.0.176.214", 16010);//服务器的连接工厂,参数为服务器ip,和连接工厂的端口号.
/**
*直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
javax.naming.Context jndiCtx = new InitialContext(pr);
// javax.naming.Context jndiCtx = new javax.naming.InitialContext(JMSConstant.getProp());
/**
*绑定对象.
*/
jndiCtx.bind("qcf", qcf);
jndiCtx.bind("queue", queue);
jndiCtx.close();//关闭上下文目录
AdminModule.disconnect();//关闭AdminModule连接
System.out.println("close the admin!");
}
}
这样就执行代码后,就在服务器上创建了用户,目的地,和对它们的jndi绑定.
三:然后执行多线程的发送消息的代码:(见6.12多线程的消息发送)
四:多线程的接收消息,代码如下.
public class Receiver1
{
QueueReceiver qrec1=null;
QueueSession qs =null;
QueueConnection qc=null;
/**
*
* @throws JMSException
* 构造函数初始化
*/
public Receiver1() throws NamingException, JMSException{
//ictx = new InitialContext(JMSConstant.getProp());
/**
*直接创建一个Properties,把jndi配置属性key,value放进去.可以省去读取jdni.properties的时间和大量代码
*/
Properties pr =new Properties();
pr.put("82.0.176.214","16400");
Context ictx = new InitialContext(pr);
Queue queue = (Queue) ictx.lookup("queue");
QueueConnectionFactory qcf = (QueueConnectionFactory) ictx.lookup("qcf");
ictx.close();
qc = qcf.createQueueConnection("user1","user1");//创建只属于user1的连接,以区别其他连接.
qs = qc.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
qrec1 = qs.createReceiver(queue,"sessionID=2");
}
/**
*
* @author Owner
* 一个内部类继承Thread实现run()方法,在run()方法设置消息监听器.
*/
class Brun extends Thread{
public void run(){
try{
qrec1.setMessageListener(new MsgListener1());
qc.start();//把连接开始应该放在此处.设置消息监听动作之后
}catch(JMSException e){}
}
}
/**
* 创建线程组,启动线程组.
*
*/
public void run_thread() {
Brun[] tt = new Brun[300];
for (int i = 0; i < tt.length; i++) {
tt[i] = new Brun();
tt[i].start();
}
}
/**
* 创建方法close()关闭消息发送器,会话,连接对象.
*
* @throws JMSException
*/
void close() throws JMSException {
qrec1.close();
qs.close();
qc.close();
}
/**
*
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception
{
System.out.println("Requests to receive messages...");
Receiver1 rc=new Receiver1();
rc.run_thread();
// qc.start(); 不能把连接开始放在此处,而是放在设置消息监听的动作之后.
System.in .read();//当消息接收完毕,在控制台上输入任意字符,执行关闭的动作.
rc.close();
System.out.println( " messages received.");
}
}
7 joram和spring的集成
7.1 简单的消息发送
一.首先启动joram服务器,管理各种对象和jndi绑定(详细请看前面章节).
二.在工程目录下引入spring支持的包和joram包.
三.在工程目录下创建applicationContext.xml文件.这个文件是spring的配置文件.示例如下
<!-----xml文件头,为默认,我们不需要修改---------------->
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">
<!------配置文件内容------->
<beans>
<!------连接工厂配置.注意:value值为jndi绑定后的key=”qcf”------->
<bean id="connectionFactory"
class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>qcf</value>
</property>
<property name="jndiTemplate">
<ref local="jndiTemplate"></ref>
</property>
</bean>
<!-----目的地配置,注意:queue为jndi绑定后的key而不是value>
<bean id="destination"
class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiName">
<value>queue</value>
</property>
<property name="jndiTemplate">
<ref local="jndiTemplate"></ref>
</property>
</bean>
<!---spring-jms模版:注意:joram是支持jms1.1的所以用org.springframework.jms.core.JmsTemplate而不是用org.springframework.jms.core.JmsTemplate102------- >
<bean id="jmsTemplate"
class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<ref local="connectionFactory" />
</property>
<property name="defaultDestination">
<ref local="destination" />
</property>
</bean>
<!-----配置jndi环境------->
<bean id="jndiTemplate"
class="org.springframework.jndi.JndiTemplate">
<property name="environment">
<props>
<prop key="java.naming.factory.initial">
fr.dyade.aaa.jndi2.client.NamingContextFactory
</prop>
<prop key="java.naming.factory.host">localhost</prop>
<prop key="java.naming.factory.port">16400</prop>
</props>
</property>
</bean>
</beans>
四.发送消息客户端代码:
public class JmsTemplateTest11 {
protected static final Log log = LogFactory.getLog(JmsTemplateTest11.class);
static Context ictx = null;
public static void main(String[] args) throws NamingException, JMSException {
System.out.println();
System.out.println("Sends messages on the queue...");
/**引入配置文件,推荐使用FileSystemXmlApplicationContext,这样你可以一次使用
*多个配置文件
*/.
ApplicationContext ac = new FileSystemXmlApplicationContext(
new String[]{
"F:\\project\\SpringJMS\\src\\applicationContext.xml"});
JmsTemplate jt = (JmsTemplate) ac.getBean("jmsTemplate");
//注入bean,由于配置文件各个bean是相互关联的,所以注入jmsTemplate就注入了//其它bean,当然你也可以这样写.但是配置文件中各个bean没有关联.这段代码可以//更深入理解spring的ioc模式
//Destination de = (Destination) ac.getBean(“destination”)
//QueueConnetionfactory cf =( QueueConnectionFactory) //ac.getBean(“connectionFactory”)
/**通过jt模版发送消息,这里new MessageCreator()使用的反射原理.
*/
jt.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
// return session
// .createTextMessage("Hello World(11),queue!");
TextMessage message = (TextMessage) session.createTextMessage("bbc");//创建一个消息,内容为bbc.
return message;
}
});
log.info("成功发送消息！");
System.out.println("Success Sends messages to the queue...");
}
}
发表评论
相关推荐
- **验证版本：** 本文档所基于的版本为 Joram 5.4.0。 **1.3 运行 Joram 服务器** - **启动命令：** 使用命令行运行 `joram-server start` 来启动 Joram 服务器。 - **配置文件：** 可以通过修改配置文件来定制...
Joram5.5作为一款面向Java的高效WebServices通信框架，不仅提供了丰富的功能模块，还通过详尽的文档和示例应用，降低了学习门槛，使得即使是初学者也能迅速掌握其核心概念与操作技巧。通过深入研究Joram的安装配置、...
jnr-unixsocket-nodep.zip,jnr:jnr unixsocket与orh.objectweb.asm shaded com.github.jnr:jnr unixsocket与orh.objectweb.asm shaded
Activiti eclipse插件。创始人Tom Baeyens是JBoss jBPM的项目架构师，以及另一位架构师Joram Barrez，一起加入到创建Alfresco这项首次实现Apache开源许可的BPMN 2.0引擎开发中来。
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用
jar包，官方版本，自测可用