友情提示：如果本网页打开太慢或显示不完整，请尝试鼠标右键“刷新”本网页！
JMS简明教程(PDF格式)-第10部分
快捷操作: 按键盘上方向键 ← 或 → 可快速上下翻页 按键盘上的 Enter 键可回到本书目录页 按键盘上方向键 ↑ 可回到本页顶部! 如果本书没有阅读完，想下次继续接着阅读，可使用上方 "收藏到我的浏览器" 功能 和 "加入书签" 功能！
小了某些消息处理排序时资源的使用。
8。2。5 ConnectionConsumer 如何使用ServerSession
由JMS 提供商实现的ConnectionConsumer 使用ServerSession 来处理一到多个到达的消
息。按以下方式做这项工作：
z 从ServerSessionPool 中得到一个ServerSession 。
z 得到ServerSession 的Session 。
z 用一个或多个消息加载Session 。
z 然后其他ServerSession 来消费这些消息。
用于QueueConnection 的ConnectionConsumer 将它的消息加载到QueueSession，同样，
用于TopicConnection 的ConnectionConsumer 加载TopicSession 。
注意，JMS 没有说明 ConnectionConsumer 如何用消息加载 Session 。由于
ConnectionConsumer 和Session 都由同一个JMS 提供商实现，因此它们可以用私有机制完成
这种加载。
8。2。6 应用服务器如何实现ServerSession
JMS 没有说明ServerSession 的实现。下面出现的普通实现解释了这个概念：
z 应用服务器为 ServerSession 创建一个线程，注册 ServerSession 的 runObject 。
runObject 的实现是应用服务器私有的。
z ServerSession 的start 方法调用线程的start 方法。和所有的java 线程一样，调用start
初始化线程的执行并调用线程的runObject。ServerSession。start（ConnectionConsumer）
的调用者和ServerSession 的runObject 现在运行在不同的线程中。
z runObject 将做一些内部处理然后调用会话的 run（）方法。在返回时，runObject 将
ServerSession 返回给ServerSessionPool 然后返回。这终止了ServerSession 线程的执
行，并再次循环启动。
8。2。7 结果
JMS 定义了灵活的机制将并发消息消费的工作拆分成适合于每个参与者的角色。
52 / 66
…………………………………………………………Page 53……………………………………………………………
应用程序员提供一种易于书写单线程的MessageListener 实现。
JMS 提供者保持对消息的控制直到它们被转发到MessageListener。这保证它是在消息确
认的直接控制之下。
应用服务器控制ConnectionConsumer 的设置和管理用于执行MessageListener 的线程。
下图解释了三个角色和它们实现的对象间的关系。
下图解释了ConnectionConsumer 将消息转发给MessageListener 的过程。
53 / 66
…………………………………………………………Page 54……………………………………………………………
8。3 XAConnectionFactory
某些应用服务器提供对用于分布式事务的资源分组提供支持。为了在分布式事务中包含
JMS 事务，应用服务器要求JTA 兼容JMS 提供商。JMS 提供商用JMS XAConnectionFactory 暴
露它的JTA 支持，应用服务器使用XAConnectionFactory 来创建XAConnection 。
XAConnectionFactory 提供了和ConnectionFactory 一样的授权选项。
XAConnectionFactory 对象是JMS 受管理对象，就像ConnectionFactory 对象。期望用应
用服务器使用JNDI 找到它们。
8。4 XAConnection
XAConnection 通过提供创建XASession 扩展了Connection 的能力。
8。5 XASession
XASession 提供了获取看起来象普通 Session 对象的对象和控制会话事务上下文的
javax。transaction。xa。XAResource 对象。XAResource 的功能非常类似于标准X/Open XA 资源接
口定义的功能。
应用服务器通过获取XAResource 来控制XASession 的事务分配。它使用XAResource 来
将会话分配给分布式事务，准备和提交事务上的工作等等。
XAResource 为多事务上的交叉工作、恢复处理中的事务列表等等提供了公平的专业工
具。JTA 感知的JMS 提供上必须全部实现这个功能。这可以通过使用支持XA 的数据库服务
来做到，或者JMS 提供商可以选择从基础开始来实现这个功能。
将XASession 的Session 赋给应用服务器的客户端。之后，应用服务器控制后台XASession
的事务管理。
但必须注意的是，分布式事务上下文不会随着消息流动；也就是说，接收消息的事务和
和生产消息的事务不能是同一个。这是异步消息和同步处理间的基本差别。消息生产者和消
费者使用两种方式来建立JMS 提供商保证一次只有一个消息被转发的能力。
再次重申，在Session 中生产和/或消费消息的行为都可以是事务性的。在不同会话间生
产和消费一个特定消息的行为不能是事务性的。
8。6 JMS 应用服务器接口
PTP 和Pub/Sub 两个域都提供了它们自己的JTS 感知的JMS 工具。
但是，应当优先使用公共接口。表8�1 列出了JMS 的公共接口。
表8�1 域内可选接口的关系
JMS 公共接口 PTP 接口 Pub/Sub 接口
ServerSessionPool 没有域特定的接口 没有域特定的接口
ServerSession 没有域特定的接口 没有域特定的接口
ConnectionConsumer 没有域特定的接口 没有域特定的接口
XAConnectionFactory XAQueueConnectionFactory XATopicConnectionFactory
54 / 66
…………………………………………………………Page 55……………………………………………………………
XAConnection XAQueueConnection XATopicConnection
XASession XAQueueSession XATopicSession
9 JMS 样例代码
本章给出了一些展示JMS 客户端如何使用JMS API 的代码样例。也解释了如何使用几个
消息类型。这些例子使用了支持统一消息模型的方法：这些例子使用点对点或发布/订阅进
行工作。用JMS API 进行工作是推荐的方式。
在JMS 规范的早期版本中，只支持消息域（点对点或发布/订阅）各自的接口，且客户
端要么使用点对点要么使用发布/订阅来进行编程。现在，JMS 客户端可以使用JMS 公共接
口进行编程。
在例子中，客户端应用发送和接收股票报价信息。客户端应用从股票报价服务中接收消
息。股票报价服务在这个例子中没有描述。
为了简化这个例子，没有对异常进行处理。
本章描述了为发送和接收消息而创建正确环境的步骤。
在描述了基本的功能后，本章描述如何执行一些其他的公共功能，例如使用消息选择器。
9。1 准备发送和接收消息
下面是建立连接然后准备发送和接收消息的基本步骤。
z 获得ConnectionFactory 和Destination。
z 创建Connection 和Session 。
z 创建MessageConsumer。
z 创建MessageProducer。
9。1。1 获取ConnectionFactory
消息生产者和消息消费者（发送者和接收者）都需要得到ConnectionFactory，并使用它
来建立Connection 和Session 。
通常，管理员已经为JMS 客户端创建和配置好了ConnectionFactory 。客户端程序通常使
用JNDI API 来查找ConnectionFactory 。
import javax。naming。*；
import javax。jms。*；
ConnectionFactory connectionFactory；
Context messaging = new InitialContext（）；
connectionFactory = （ConnectionFactory）
messaging。lookup（〃ConnectionFactory〃）；
55 / 66
…………………………………………………………Page 56……………………………………………………………
9。1。2 获取Destination
管理员已经创建和配置了一个名字为“StockQueue ”的Queue，它是股票报价消息被发
送和接收的地方。同样，可以使用JNDI API 来查找目的地。
Queue stockQueue；
stockQueue = （Queue）messaging。lookup（〃StockSource〃）；
9。1。3 创建Connection
在得到ConnectionFactory 之后，客户端程序使用它来创建Connection 。
Connection connection；
connection = ConnectionFactory。createConnection（）；
9。1。4 创建Session
在得到 Connection 之后，客户端程序使用它来创建 Session 。这个 Session 用于创建
MessageProducer （发送消息）或MessageConuser （接收消息）。
Connection。createSession 方法有两个参数：
z 一个表示会话是否是事务性的布尔。
z 消息确认模式。
Session session；
/* Session is not transacted；
* uses AUTO_ACKNOWLEDGE for message
* acknowledgement
*/
session = connection。createSession（false； Session。AUTO_ACKNOWLEDGE）；
9。1。5 创建MessageProducer
在得到 Session 后，客户端程序使用 Session 来创建 MessageProducer 。这个
MessageProducer 对象用于向目的地发送消息。通过使用Session。createProducer 方法来创建
MessageProducer，参数是消息被发送的目的地。
MessageProducer sender；
/* Value in stockQueue previously looked up in the JNDI
* createProducer takes a Destination
*/
sender = session。createProducer（stockQueue）；
9。1。6 创建MessageConsumer
消息可以被同步或异步的消费。这个例子展示了如何创建同步消费消息的消息消费者。
56 / 66
…………………………………………………………Page 57……………………………………………………………
参见9。3。1 “异步接收消息”了解如何异步消费消息。
MessageConsumer 用于从目的地接收消息，在这个例子中是“StockQueue ”。用
Session。createConsumer 方法来创建MessageConsumer，参数是接收消息的目的地。
MessageConsumer receiver；
/* Value in stockQueue previously looked up in the JNDI
* createConsumer takes a Destination
*/
receiver = session。createConsumer（stockQueue）；
9。1。7 启动消息转发
到现在为止，消息的转发还是被禁止的，以便前面的设置不会被异步转发的消息打断。
现在设置已经完成，可以告诉Connection 开始向MessageConsumer 转发消息了。
connection。start（）；
9。1。8 使用TextMessage
有几个消息格式。在这个例子中，股票报价消息作为文本串被发送。客户端按文本串读
取和展示这个消息。
下面的代码解释了如何创建这样的消息：
String stockData； /* Stock information as a string */
TextMessage message；
/* Set the message’s text to be the stockData string */
message = session。createTextMessage（）；
message。setText（stockData）；
9。2 发送和接收消息
现在已经完成了Session 的设置，你可以发送和接收消息了。本节描述了如何：
z 创建消息
z 发送消息
z 同步接收消息
9。2。1 发送消息
为了发送消息，使用MessageProducer。send 方法，参数是Message 对象。
/* Send the message */
sender。send（message）；
57 / 66
…………………………………………………………Page 58……………………………………………………………
9。2。2 同步接收消息
为了接收Queue 里的消息，你可以使用MessageConsumer。receive 方法。这个方法调用
在消息到达Queue 之前可能阻塞。这个方法也可以用于从Topic 中接收消息。
TextMessage stockMessage；
stockMessage = （TextMessage）receiver。receive（）；
为了限制客户端阻塞的时间长度，使用 receive 方法的超时参数。如果在超时到期时还
没有消息到达，那么receive 方法返回。超时参数按毫秒计时。
TextMessage stockMessage；
/* Wait 4 seconds for a message */
TextMessage = （TextMessage）receiver。receive（4000）；
9。2。3 解包TextMessage
股票报价信息使用 TextMessage 发送。为了从消息中得到这个信息，使用
TextMessage。getText 方法。它以字符串返回消息的内容。
String newStockData； /* Stock information as a string */
newStockData = message。getText（）；
9。3 其他消息特性
本节不仅提供了基本的消息功能，而且还描述了如何执行某些其他的公共消息功能：
z 创建异步MessageListener。
z 使用消息选择器过滤消息转发。
z 创建Topic 的永久订阅。
z 对使用永久订阅的Topic 进行重新连接。
9。3。1 异步接收消息
为了在消息被转发到消息消费者时异步的接收消息，客户端程序需要创建实现了
MessageListener 接口的消息监听器。本例中 MessageListener 接口的实现称为
StockListener。java，类似于下面的代码：
import javax。jms。*；
public class StockListener implements MessageListener
｛
public void onMessage（Message message） ｛
/* Unpack and handle the messages received */
。。。
｝
｝
客户端程序按下面的方式将这个MessageListener 对象注册到MessageConsumer 对象：
58 / 66
…………………………………………………………Page 59……………………………………………………………
StockListener myListener = new StockListener（）；
/* Receiver is MessageConsumer object */
receiver。setMessageListener（myListener）；
Connection 必须被启动以迎接开始的消息转发。当消息被发布到 Queue 时会异步的通
知MessageListener。这通过MessageListener 接口的onMessage 方法实现。由客户端负责处
理消息。
public void onMessage（Message message）
｛
String newStockData；
/* Unpack and handle the messages received */
newStockData = message。getText（）；
if（。。。）
｛
/* Logic related to the data */
｝
｝
9。3。2 使用消息选择器
在这个例子中，客户端程序只对技术相关的股票感兴趣。消息的发送者向消息的属性中
设置一个称为StockSector 的属性。它的值包括“Technology ”，“Financial”，“Manufacturing”；
“Emerging”和“Global ”。消息发送者通过使用Message。setStringProperty 方法来设置这些
属性值。
String stockData； /* Stock information as a String */
TextMessage message；
/* Set the message’s text to be the stockData string *
message = session。createTextMessage（）；
message。setText（stockData）；
/* Set the message property ‘StockSector’
message。setStringProperty（〃StockSector〃； 〃Technology〃）；
当接收股票报价消息客户端程序 MessageConsumer 被创建时，它可以创建一个消息选
择器字符串来决定它将接收哪些消息。
String selector；
selector = new String（〃（StockSector = ‘Technology’）〃
这个字符串在创建MessageConsumer 时指定。
MessageConsumer receiver；
receiver = session。createConsumer（stockQueue； selector）；
客户端程序只接收和Technology 相关的消息。
9。3。3 使用永久订阅
永久订阅用于从Topic 接收消息。当JMS 客户端创建永久订阅时，客户端能够从Topic
断开连接。当客户端程序重新连接时，它可以收到断开连接后到达的消息。在这个例子中，
59 / 66
…………………………………………………………Page 60……………………………………………………………
Destination 提供对消息更新的通知。
9。3。3。1 创建永久订阅
下面的例子创建了从Topic 获取消息的永久订阅。首先，客户端程序必须执行常规的设
置步骤，查找ConnectionFactory 和Destination，创建Connection 和Session，如节9。1 “准备
发送和接收消息”中所述。
im
快捷操作: 按键盘上方向键 ← 或 → 可快速上下翻页 按键盘上的 Enter 键可回到本书目录页 按键盘上方向键 ↑ 可回到本页顶部!
温馨提示： 温看小说的同时发表评论，说出自己的看法和其它小伙伴们分享也不错哦！发表书评还可以获得积分和经验奖励，认真写原创书评 被采纳为精评可以获得大量金币、积分和经验奖励哦！