基于JMS的数据汇集系统的研究与实现

摘要 在深入研究JMS异步消息处理机制的基础上，利用JMS为分布在不同系统中的气象数据的汇集提供了一个可行的方案，解开了应用程序间的耦合，并增强了应用程序的可扩展性和可靠性。

　　关键词 消息、JMS、点对点、发布/订阅、数据汇集、气象

　　现如今，越来越多的企业、单位面临着各种各样的数据集成和系统整合，CORBA、DCOM、RMI等RPC中间件技术也应运而生，但由于采用RPC同步处理技术，在性能、健壮性、可扩展性上都存在着诸多缺点。而基于消息的异步处理模型采用非阻塞的调用特性，发送者将消息发送给消息服务器，消息服务器在合适的时候再将消息转发给接收者；发送和接收是异步的，发送者无需等待，二者的生命周期也可以不必相同，而且发送者可以将消息间接传给多个接收者，大大提高了程序的性能、可扩展性及健壮性，这使得异步处理模型在分布式应用上比起同步处理模型更具有吸引力。

　　本文详细介绍了Java中的异步处理机制――基于JMS的异步消息处理技术，并结合实例讨论了基于JMS的气象系统数据汇集系统的设计与实现。

　　Java消息服务－JMS

　　JMS是由包括Sun Microsystems和IBM等在内的几个公司合作设计的一个面向消息的中间件（MOM）API。JMS定义了Java 中访问消息中间件的接口，但JMS 只是接口，并没有给予实现，实现JMS 接口的消息中间件称为JMS提供者（JMS Provider）。JMS的目的是应用程序能在异步情况下可靠地传输和接受消息。从编程的角度来看，JMS可以被当作一个容器管理的资源，与JDBC连接类似。正如你可以通过JDBC去访问许多不同的关系数据库一样，你可以通过JMS访问独立于厂商的消息服务系统。

　　JMS主要包括三个部分――JMS提供者（JMS Provider）、消息发送者（Message Producer）和消息接收者（Message Consumer）。简单的说，JMS提供者是指实现JMS API接口的消息系统，是消息的中转站；消息发送者是指消息的创造和发送者，是消息的发源地；消息接收者是指接收消息的应用程序，是消息最终的目的地。

　　JMS消息是异步处理的，消息发送者可以发送一个消息而无须等待响应。消息发送者将消息发送到一条虚拟的通道(主题或队列)上，消息接收者则订阅或是监听该通道。一条消息可能最终转发给一个或多个消息接收者，这些接收者都无需对消息发送者做出回应。

　　1、JMS消息的组成

　　在J2EE1.4后，JMS API不再区分在点对点域和发布/订阅域中创建的消息。JMS消息由以下三部分组成（如图1.1所示）：

　　消息头（header）――JMS消息头包含了许多字段，它们是消息发送后由JMS提供者或消息发送者产生，用来表示消息、设置优先权和失效时间等等，并且为消息确定路由。

　　属性（property）――由消息发送者产生，用来添加删除消息头以外的附加信息。

　　消息体（body）――由消息发送者产生，JMS中定义了5种消息体：ByteMessage、MapMessage、ObjectMessage、StreamMessage和TextMessage。



　　2、JMS消息传递模式

　　JMS支持两种消息传递模式――点对点模式（P2P）和发布/订阅模式（Publish/Subscribe）。这两种都是人们熟知的push模式，消息的发送者是活动的发起人，而接收者则是被动的接收消息。在JMS中，这些消息传递模式被称为消息域（message domain）。

　　2.1 JMS点对点消息域

　　在点对点模式中，发送者和接收者对消息传送的目的地址达成一致，即所谓的队列（queue）。消息队列位于JMS提供者中，消息发送者向一个消息队列发送消息，消息接收者可以在消息发送后的任何时刻从这个队列中（被动地）接收消息，在接收者确认之前消息一直保存在消息队列中直到过期。点对点模式的结构图如图1.2所示：

　　JMS点对点消息域具有以下特点：

　　·每条消息能而且只能被一个接收者接收。

　　·每条消息或者被接收者从队列中取走，或者被 JMS提供者在超时的情况下删除。

　　·消息产生的时候接收者不一定要存在，接收者可以在消息产生后的任何时间里取走消息。

　　·接收者不能请求一个消息。

　　·接收者必须在收到消息后发出确认信息。



　　2.2 JMS发布/订阅消息域

　　在发布/订阅模式下，发送者被称为发布者（publisher），一个消息可已有很多接收者，这些接收者被称为订阅者（subscriber）。发布/订阅模式采用与点对点模式完全不同的消息发送模式。在发布/订阅模式下，发布者给一个主题（topic）发送消息，多个订阅者在订阅的时候可以订阅他们感兴趣的主题。一个主题可以被多个订阅者订阅，一个订阅者也可以订阅多个主题。一个主题的消息只被发给该主题的所有订阅者。订阅者只能接收它订阅的主题中的消息，并且，在默认情况下，订阅者在消息发送的时候必须是活动的，并随时准备接收消息，否则它将错过该消息。为了避免这种时间依赖性JMS API允许订阅者创建持久订阅。发布/订阅模式的结构图如图1.3所示：

　　JMS发布/订阅消息域具有以下特点：

　　·每一条消息由一个发布者创建而由0个或多个订阅者接收它。

　　·消息立刻被分发给现有的订阅者。

　　·订阅者必须在消息发送的时候存在以接收消息。

　　·持久订阅允许订阅者接收它处于非活动状态时由发布者向主题发送的消息。

　　·订阅者必须在接收到消息后发出确认信息。

　　

系统的设计与实现

　　目前，基层气象台站的每个分系统都是独立的，各自的数据都保存在本机上，彼此互不联系，而汇集系统就是为了把分散在每个系统上的数据汇集到一个统一的数据库中。为了实现这个目的可以有多种方法。

　　（1）由各个分系统解析本机的数据，然后直接向数据库中写记录，但是这样势必对统一数据库的安全性造成影响，而且各分系统的计算机莨莠不齐，老旧的机器能不能提供更多的资源也是一个问题；

　　（2）由服务器上的一个程序统一从各个系统上获取数据，然后写入数据库，但是这样该服务器程序与各个分系统便形成了紧耦合，一旦分系统发生变化，或者要扩展一个分系统便要重新更改程序代码，使得整个系统的稳定性和可扩展性受到很大影响；

　　而JMS消息机制的异步传输模型恰好能完美的解决以上两种方法存在的问题。

　　1、系统的设计

　　基于JMS的气象数据汇集系统由五个模块组成：代理模块、消息发送模块（MessageSender）、消息接收模块（MessageReceiver）、数据格式转换模块（DataFormatConverser）和统一数据库访问接口（UDAI）。系统结构图如下图所示：

 

 

基于JMS的气象数据汇集系统结构图

　　（1）代理模块

　　代理模块是运行于分系统上的Java程序，它只有两个功能：监测到有新数据到，立即将数据文件复制到服务器上的文件缓冲区，然后调用消息发送模块，并将数据文件的文件名，文件大小，生成时间，业务属性等信息传递给消息发送模块。由于代理程序不需要解析数据文件，也不需要写数据库，降低了对系统资源的消耗，即使在比较老旧的机器上运行也不会出现问题。而且一旦消息发送成功，后面的工作便与分系统无关，这样也就解除了分系统和服务器之间的耦合。

　　（2）消息发送模块

　　消息发送模块作为一个会话Bean部署在服务器上，它的功能主要是：将代理模块传来的数据文件的文件名，文件大小，生成时间，业务属性等信息组装成消息，然后发送到与其业务属性相对应的消息队列中（每一类业务属性对应一个消息队列）。消息发送模块运行与服务器上的Bean容器中，不占用分系统资源，减小了分系统上的资源开销；而且该模块被部署为一个Bean，方便了以后分系统的扩展和变化。

　　（3）消息接收模块

　　消息接收模块是一个运行于服务器端的JMS客户端，每一个消息队列都对应一个消息接收模块。消息接收模块通过注册一个MessageListener接口，监听消息队列上的消息；接收到消息后便调用数据格式转换模块，将消息内容和数据文件在文件缓冲区的绝对地址作为参数传给数据格式转换模块。之所以为每一个消息队列对应一个消息接收模块，是考虑到以后系统的扩展，如果要增加一个队列，只需增加一个接收模块，并不影响原来系统的正常运行。

 

　　（4）数据格式转换模块

　　数据格式转换模块的主要功能是：解析数据文件并按照一定的格式生成标准的XML文档供统一数据库访问接口使用。

　　（5）统一数据库访问接口

　　统一数据库访问接口提供一个统一的数据库写入接口，不管以后系统如何变化，只要生成标准的XML文档便可以使用此接口来访问数据库。

　　系统中真正与JMS相关的模块是消息发送模块和消息接收模块，由于每一类业务属性对应一个消息队列，每一个消息队列对应一个接收模块，所以选择PTP的传递模式作为系统的消息传递模式。

　　2、系统的实现

　　（1）建立消息队列

　　系统使用JBoss作为J2EE服务器，通过编辑配置文件jbossMQ-destinations-service.xml来定义应用所需的消息队列地址信息。其部分代码如下：

＜mbean code="org.jboss.mq.server.jmx.Queue" name="jboss.mq.destination:service=Queue,name=RadarQueue"＞
＜depends optional-attribute-name="DestinationManager"＞
jboss.mq:service=DestinationManager＜/depends＞

＜depends optional-attribute-name="SecurityManager"＞
jboss.mq:service=SecurityManager＜/depends＞
＜attribute name="SecurityConf"＞
＜security＞
＜role name="guest" read="true" write="true"/＞
＜role name="publisher" read="true" write="true" create="false"/＞
＜role name="noacc" read="false" write="false" create="false"/＞
＜/security＞

＜/attribute＞
＜/mbean＞
　　（2）消息发送模块的实现

　　MessageSender SessionBean的功能是构造消息并向指定的消息队列发送消息，其部分代码如下：

//构造TEXT消息

message.setText(dataInfo);//设置消息体

//发送消息

Context context = new InitiaContext();//获取上下文环境
ConnectionFactory connectionFactory = new (ConnectionFactory)context.lookup(“java:comp/env/JNDIConnectionFactory”);

Connection connection = connectionFactory.createConnection();
Session session =connection.createSession(false,AUTO_ACKNOWLEDGE);
Destination messageQueue = (Destination)context.lookup(“java:comp/env/ RadarQueue”);
MessageProducer producer = session.createProducer(messageQueue);
producer.send(message);
……
　　（3）消息接收模块的实现

　　消息接收模块通过注册MessageListener接口来监听消息队列，实现其onMessage()方法处理消息。其部分代码如下：

Public class MessageReceiver implements MessageListener{
……
public void onMessage(Message message){
……
dataInfo = ((TextMessage)message).getText();//获得数据文件信息
……
//调用数据格式转换模块
……
}
　　结束语

　　本系统应用JMS异步消息作为分系统与服务器的关联机制，分系统只需要通过消息向服务器提交任务，之后便不需要关心数据如何处理，何时处理，彻底解开了分系统与服务其的耦合，即保证了数据库的安全可靠，也减少了分系统的资源消耗。而且系统的各个模块都是相对独立，对于今后的变化和进一步的扩展提供了更大的空间。