基于JMS的數(shù)據(jù)清分

1、基于JMS的數(shù)據(jù)清分1 JMS概述JMS是一個標(biāo)準(zhǔn)的應(yīng)用編程接口API，用來建立廠商中立的異步通信機制。從這個意義上說，JMS類似于JDBC和JNDI，例如就JDBC來說，它要求有一個底層的數(shù)據(jù)庫提供者，JMS則要求有一個支持JMS標(biāo)準(zhǔn)的底層異步通信中間件提供者，一般稱為面向消息的中間件Message-Oriented Middleware，MOM。 MOM是一種支持應(yīng)用程序通過交換消息實現(xiàn)異步通信的技術(shù)。在某種程度上，異步通信有點象是人們通過email進(jìn)行通信；類似地，同步通信的程序就象是人們通過 進(jìn)行通信。在異步通信過程中，程序的結(jié)合方式是寬松的，換句話說，異步通信的程序并不直接相互聯(lián)系，

2、而是通過稱為隊列Queue或主題Topic的虛擬通道聯(lián)系。JMS有兩種通信方式：第一種是端對端通信Point-to-Point，P2P方式，第二種是出版/訂閱Publish/Subscribe，Pub/Sub方式。1.1 P2P方式P2P消息傳遞又可以按照推Push和拉Pull兩種方式運作。在P2P拉方式中，程序通過稱為隊列的虛擬通道通信：在通信會話的發(fā)送方，發(fā)送程序把一個消息"放入"隊列，在接收方，接收程序定期掃描隊列，尋找它希望接收和處理的消息。和推方式相比，拉方式的消息傳遞效率較低，因為它需要周而復(fù)始地檢查消息是否到達(dá)，這個過程會消耗一定的資源。另外必須注意的一點是，

3、當(dāng)接收方發(fā)現(xiàn)一個需要處理的消息時，它就會"提取"消息，從效果上看等于從隊列刪除了消息。 因此，即使有多個接收程序在處理同一個隊列，對于某一特定的消息來說，總是只有一個接收者。JMS程序可以使用多個隊列，每一個隊列可以由多個程序處理，但是只有一個程序才會收到某個特定的消息。 在P2P推方式的消息傳遞中，發(fā)送程序的工作原理也一樣，它把消息發(fā)送到隊列，但接收程序的工作原理有所不同。接收程序?qū)崿F(xiàn)了一個Listener接口，包括實現(xiàn)了該接口中的onMessage回調(diào)方法，在J2EE環(huán)境中監(jiān)聽隊列接收消息的任務(wù)交給了容器，每當(dāng)一個新的消息到達(dá)隊列，容器就調(diào)用onMessage方法，將消

4、息作為參數(shù)傳遞給onMessage方法。 P2P通信最重要的特點不管是推還是拉是：每一個消息總是只由一個程序接收。一般而言，P2P程序在通信過程中參與的活動較多，例如，發(fā)送程序可以向接收程序指出應(yīng)答消息應(yīng)當(dāng)放入哪一個隊列，它還可以要求返回一個確認(rèn)或報告消息。1.2 Pub/Sub方式在Pub/Sub通信方式中，程序之間通過一個主題Topic實現(xiàn)通信。在消息發(fā)送方，生產(chǎn)消息的程序向主題發(fā)送消息；在接收方，消息的消費程序向感興趣的主題訂閱消息。當(dāng)一個消息到達(dá)主題，所有向該主題訂閱的消費程序都會通過onMessage方法的參數(shù)收到消息。這是一種推式的消息傳遞機制?？梢栽O(shè)想，會有一個以上的消費程序收到

5、同一消息的副本。相比之下，程序在Pub/Sub通信過程中參與的活動較少，當(dāng)生產(chǎn)者程序向某個特定的隊列發(fā)送消息，它不知道到底會有多少程序接收到該消息可能有多個，可能沒有。2 數(shù)據(jù)清分2.1 清分模式清分程序從總公司把數(shù)據(jù)按各分公司分類，分別發(fā)送給各個分公司，分公司接收處理數(shù)據(jù)后反饋給總公司，如下列圖：總公司分公司A分公司B分公司N在數(shù)據(jù)清分中，對于某一筆數(shù)據(jù)來說，只發(fā)送給某個地方，也就是說一個消息只有一個消費者；從這個意義上來說P2P方式更適合做數(shù)據(jù)清分。如果數(shù)據(jù)是單向的傳輸，即只從總公司發(fā)送給分公司，那么只需要每個公司一個隊列就可以實現(xiàn)，總公司配置“反饋隊列”，分公司配置“消息隊列”，如下列圖

6、：總公司分公司反饋隊列消息隊列如果數(shù)據(jù)是雙向的，即分公司也要發(fā)送數(shù)據(jù)給總公司，那么每個公司都需要配置兩個隊列，如下列圖：總公司分公司反饋隊列消息隊列消息隊列反饋隊列我們的實現(xiàn)的程序支持雙向方式，我們稱之為ssjms，后面文檔中的部署例子是雙向方式的。2.2 ssjms介紹ssjms實現(xiàn)了雙向消息發(fā)送機制，程序包中提供了發(fā)送消息的API，消息的接受采用MDB消息EJB來自動處理，下列圖是文件的結(jié)構(gòu)。src是源代碼目錄，resource是ejb的配置文件，test是發(fā)送消息的測試?yán)印?.2.1 發(fā)送消息com.sinosoft.jms.core.Producer類提供了發(fā)送消息方法sendMes

7、sage。以下是發(fā)送消息的例子，在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、);下面講下發(fā)送消息的步驟13：1、 創(chuàng)建Producer的對象Producer producer = new Producer("t3:/localhost:5050","t3:/localhost:6060");構(gòu)造函數(shù)的入?yún)⒑x如下：/* * 消息生產(chǎn)者 * param fromServerURL 發(fā)送端服務(wù)器URL * param toServerURL 接收端服務(wù)器URL */public Producer(String fromServerURL, String toServerURL)2、 創(chuàng)建消息體MessageBodyMessageBo

11、dy body = new MessageBody();消息體可以通過setData和setProperties方法來設(shè)置消息數(shù)據(jù)，3、 把消息發(fā)送出去String id = producer.sendMessage(body);調(diào)用Producer.sendMessage方法，把消息發(fā)送出去。2.2.2 接收消息我們采用了MDB自動接收消息，分成兩種MDB：接收消息EJB和接收反饋EJB；由于我們是雙向的方式，因此需要實現(xiàn)兩對EJB4個EJB，分別是：分公司接收消息EJB總公司接收反饋EJB，總公司接收消息EJB分公司接收反饋EJB。 接收消息EJB新編寫類，繼承Request

12、ConsumerBean類，覆蓋方法receiveMessage：public MessageBody receiveMessage(MessageBody messageBody);入?yún)⑹墙邮盏降南?，返回需要反饋的消息。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內(nèi)容需要根據(jù)繼承的類名修改。 接收反饋EJB新編寫類，繼承ResponseConsumerBean類，覆蓋方法receiveMessage：public MessageBody receiveMessage(MessageBody messageBody);入?yún)⑹墙邮盏降姆答佅?，返回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內(nèi)容需要根據(jù)繼承的類名修改。2.2.3 部署到weblogic上在部署之前，我們需要準(zhǔn)備好兩對EJB，總公司的2個EJB打成一個jar包，分公司的2個EJB也打成一個jar包?？偣竞头止镜牟渴鸩襟E實際一樣，只是用的EJB-jar包不同而已。我們以分公司為例列出部署步驟。1、 創(chuàng)建JMS Connection FactoryServices->JMS->Connection Factories-> Configure a new JMS Connection Factory.其中JNDIName的框填“queueCo

18、nnectionFactory”，如下列圖：創(chuàng)建后，千萬別忘了指定Targets。2、 創(chuàng)建StoreServices->JMS->Stores-> Configure a new JMSFile Store.其中Directory的框填個目錄路徑，持久化消息用，如下列圖：3、 創(chuàng)建JMS ServerServices->JMS->Servers-> Configure a new JMSServer.其中Store選擇剛剛2步創(chuàng)建的，如下列圖：創(chuàng)建后，千萬別忘了指定Targets。4、 創(chuàng)建QueueServices->JMS->Servers-> MyJMSServer-> Destinations-> Configure a new JMSQueue.創(chuàng)建兩個隊列，messageQueue和replyQueue，其中JNDIName分別填“messageQueue”和“replyQueue”，如下列圖：5、 發(fā)布EJBDeployments->EJB-> Configure a new EJB.選擇準(zhǔn)備好的jar包，指定Targets，部署后如下列圖：以