基于JMS的数据清分

1、基于JMS的数据清分1 JMS概述JMS是一个标准的应用编程接口API，用来建立厂商中立的异步通信机制。从这个意义上说，JMS类似于JDBC和JNDI，例如就JDBC来说，它要求有一个底层的数据库提供者，JMS则要求有一个支持JMS标准的底层异步通信中间件提供者，一般称为面向消息的中间件Message-Oriented Middleware，MOM。 MOM是一种支持应用程序通过交换消息实现异步通信的技术。在某种程度上，异步通信有点象是人们通过email进行通信；类似地，同步通信的程序就象是人们通过 进行通信。在异步通信过程中，程序的结合方式是宽松的，换句话说，异步通信的程序并不直接相互联系，

2、而是通过称为队列Queue或主题Topic的虚拟通道联系。JMS有两种通信方式：第一种是端对端通信Point-to-Point，P2P方式，第二种是出版/订阅Publish/Subscribe，Pub/Sub方式。1.1 P2P方式P2P消息传递又可以按照推Push和拉Pull两种方式运作。在P2P拉方式中，程序通过称为队列的虚拟通道通信：在通信会话的发送方，发送程序把一个消息"放入"队列，在接收方，接收程序定期扫描队列，寻找它希望接收和处理的消息。和推方式相比，拉方式的消息传递效率较低，因为它需要周而复始地检查消息是否到达，这个过程会消耗一定的资源。另外必须注意的一点是，

3、当接收方发现一个需要处理的消息时，它就会"提取"消息，从效果上看等于从队列删除了消息。 因此，即使有多个接收程序在处理同一个队列，对于某一特定的消息来说，总是只有一个接收者。JMS程序可以使用多个队列，每一个队列可以由多个程序处理，但是只有一个程序才会收到某个特定的消息。 在P2P推方式的消息传递中，发送程序的工作原理也一样，它把消息发送到队列，但接收程序的工作原理有所不同。接收程序实现了一个Listener接口，包括实现了该接口中的onMessage回调方法，在J2EE环境中监听队列接收消息的任务交给了容器，每当一个新的消息到达队列，容器就调用onMessage方法，将消

4、息作为参数传递给onMessage方法。 P2P通信最重要的特点不管是推还是拉是：每一个消息总是只由一个程序接收。一般而言，P2P程序在通信过程中参与的活动较多，例如，发送程序可以向接收程序指出应答消息应当放入哪一个队列，它还可以要求返回一个确认或报告消息。1.2 Pub/Sub方式在Pub/Sub通信方式中，程序之间通过一个主题Topic实现通信。在消息发送方，生产消息的程序向主题发送消息；在接收方，消息的消费程序向感兴趣的主题订阅消息。当一个消息到达主题，所有向该主题订阅的消费程序都会通过onMessage方法的参数收到消息。这是一种推式的消息传递机制。可以设想，会有一个以上的消费程序收到

5、同一消息的副本。相比之下，程序在Pub/Sub通信过程中参与的活动较少，当生产者程序向某个特定的队列发送消息，它不知道到底会有多少程序接收到该消息可能有多个，可能没有。2 数据清分2.1 清分模式清分程序从总公司把数据按各分公司分类，分别发送给各个分公司，分公司接收处理数据后反馈给总公司，如下列图：总公司分公司A分公司B分公司N在数据清分中，对于某一笔数据来说，只发送给某个地方，也就是说一个消息只有一个消费者；从这个意义上来说P2P方式更适合做数据清分。如果数据是单向的传输，即只从总公司发送给分公司，那么只需要每个公司一个队列就可以实现，总公司配置“反馈队列”，分公司配置“消息队列”，如下列图

6、：总公司分公司反馈队列消息队列如果数据是双向的，即分公司也要发送数据给总公司，那么每个公司都需要配置两个队列，如下列图：总公司分公司反馈队列消息队列消息队列反馈队列我们的实现的程序支持双向方式，我们称之为ssjms，后面文档中的部署例子是双向方式的。2.2 ssjms介绍ssjms实现了双向消息发送机制，程序包中提供了发送消息的API，消息的接受采用MDB消息EJB来自动处理，下列图是文件的结构。src是源代码目录，resource是ejb的配置文件，test是发送消息的测试例子。2.2.1 发送消息com.sinosoft.jms.core.Producer类提供了发送消息方法sendMes

7、sage。以下是发送消息的例子，在test目录中可以找到。package com.sinosoft.jms.core.examples;import java.util.Properties;import javax.jms.JMSException;import javax.naming.NamingException;import com.sinosoft.jms.core.Log;import com.sinosoft.jms.core.MessageBody;import com.sinosoft.jms.core.Producer;public class ProducerTest

8、/* * param args * throws JMSException * throws NamingException */public static void main(String args) throws NamingException, JMSException Producer producer = new Producer("t3:/localhost:5050","t3:/localhost:6060");MessageBody body = new MessageBody();body.setData("hello wor

9、ld!");Properties properties = new Properties();properties.setProperty("BizType", "Policy");properties.setProperty("PK", "PDAA200800000000000001");body.setProperties(properties);String id = producer.sendMessage(body);L("JMSMessageID = "+id

10、);下面讲下发送消息的步骤13：1、 创建Producer的对象Producer producer = new Producer("t3:/localhost:5050","t3:/localhost:6060");构造函数的入参含义如下：/* * 消息生产者 * param fromServerURL 发送端服务器URL * param toServerURL 接收端服务器URL */public Producer(String fromServerURL, String toServerURL)2、 创建消息体MessageBodyMessageBo

11、dy body = new MessageBody();消息体可以通过setData和setProperties方法来设置消息数据，3、 把消息发送出去String id = producer.sendMessage(body);调用Producer.sendMessage方法，把消息发送出去。2.2.2 接收消息我们采用了MDB自动接收消息，分成两种MDB：接收消息EJB和接收反馈EJB；由于我们是双向的方式，因此需要实现两对EJB4个EJB，分别是：分公司接收消息EJB总公司接收反馈EJB，总公司接收消息EJB分公司接收反馈EJB。 接收消息EJB新编写类，继承Request

12、ConsumerBean类，覆盖方法receiveMessage：public MessageBody receiveMessage(MessageBody messageBody);入参是接收到的消息，返回需要反馈的消息。META-INF/ejb-jar.xml：<!DOCTYPE ejb-jar PUBLIC "-/Sun Microsystems, Inc./DTD Enterprise JavaBeans 2.0/EN" " :/java.sun /dtd/ejb-jar_2_0.dtd"><ejb-jar> <en

13、terprise-beans> <message-driven> <ejb-name>RequestConsumerBean</ejb-name> <ejb-class>com.sinosoft.jms.mdb.RequestConsumerBean</ejb-class> <transaction-type>Container</transaction-type> <message-driven-destination> <destination-type>javax.jms.Q

14、ueue</destination-type> </message-driven-destination> </message-driven> </enterprise-beans></ejb-jar>其中ejb-class内容需要根据继承的类名修改。 接收反馈EJB新编写类，继承ResponseConsumerBean类，覆盖方法receiveMessage：public MessageBody receiveMessage(MessageBody messageBody);入参是接收到的反馈消息，返回null。MET

15、A-INF/ejb-jar.xml：<!DOCTYPE ejb-jar PUBLIC "-/Sun Microsystems, Inc./DTD Enterprise JavaBeans 2.0/EN" " :/java.sun /dtd/ejb-jar_2_0.dtd"><ejb-jar> <enterprise-beans> <message-driven> <ejb-name>ResponseConsumerBean</ejb-name> <ejb-class>co

16、m.sinosoft.jms.mdb.ResponseConsumerBean</ejb-class> <transaction-type>Container</transaction-type> <message-driven-destination> <destination-type>javax.jms.Queue</destination-type> </message-driven-destination> </message-driven> </enterprise-beans&g

17、t;</ejb-jar>其中ejb-class内容需要根据继承的类名修改。2.2.3 部署到weblogic上在部署之前，我们需要准备好两对EJB，总公司的2个EJB打成一个jar包，分公司的2个EJB也打成一个jar包。总公司和分公司的部署步骤实际一样，只是用的EJB-jar包不同而已。我们以分公司为例列出部署步骤。1、 创建JMS Connection FactoryServices->JMS->Connection Factories-> Configure a new JMS Connection Factory.其中JNDIName的框填“queueCo

18、nnectionFactory”，如下列图：创建后，千万别忘了指定Targets。2、 创建StoreServices->JMS->Stores-> Configure a new JMSFile Store.其中Directory的框填个目录路径，持久化消息用，如下列图：3、 创建JMS ServerServices->JMS->Servers-> Configure a new JMSServer.其中Store选择刚刚2步创建的，如下列图：创建后，千万别忘了指定Targets。4、 创建QueueServices->JMS->Servers-> MyJMSServer-> Destinations-> Configure a new JMSQueue.创建两个队列，messageQueue和replyQueue，其中JNDIName分别填“messageQueue”和“replyQueue”，如下列图：5、 发布EJBDeployments->EJB-> Configure a new EJB.选择准备好的jar包，指定Targets，部署后如下列图：以