1. JMS基本概念
JMS(Java Message Service)是訪問企業(yè)消息系統(tǒng)的標(biāo)準(zhǔn)API,它便于消息系
統(tǒng)中的Java應(yīng)用程序進(jìn)行消息交換,并且通過提供標(biāo)準(zhǔn)的產(chǎn)生、發(fā)送、接收消息的接口簡(jiǎn)化企業(yè)應(yīng)用的開發(fā)。
2. JMS基本功能
JMS是用于和面向消息的中間件相互通信的應(yīng)用程序接口。它既支持點(diǎn)對(duì)點(diǎn)(point-to-point)的域，又支持發(fā)布/訂閱(publish/subscribe)類型的域，并且提供對(duì)下列類型的支持：經(jīng)認(rèn)可的消息傳遞,事務(wù)型消息的傳遞，一致性消息和具有持久性的訂閱者支持。JMS還提供了另一種方式來對(duì)您的應(yīng)用與舊的后臺(tái)系統(tǒng)相集成。
3. WebLogic JMS Server介紹
WebLogic Server8.1符合JAVA規(guī)范,并通過Sun Microsystems J2EE 1.3認(rèn)
證.作為WebLogic的一部分,當(dāng)然WebLogic JMS Server也完全遵從JMS規(guī)范,還支持集群,并可以應(yīng)用于實(shí)際企業(yè)系統(tǒng).下圖是WebLogic JMS Server體系結(jié)構(gòu).圖中可以看到WebLogic JMS Server主要組件有: WebLogic JMS servers(用于消息通信),Java客戶端,JNDI(用于域名查找), 后備存儲(chǔ)(用于持久消息存儲(chǔ),基于文件或者JDBC數(shù)據(jù)庫).
二. WebLogic JMS特性
1. 消息通信模型
JMS 支持兩種消息通信模型：點(diǎn)到點(diǎn)（point-to-point）（PTP）模型和發(fā)布/訂閱（Pub/Sub）模型。除了下列不同之外，這兩種消息通信模型非常地相似：
PTP 模型規(guī)定了一個(gè)消息只能有一個(gè)接收者;Pub/Sub 模型允許一個(gè)消息可以有多個(gè)接收者。
2. 消息組成
消息傳遞系統(tǒng)的中心就是消息。
一條 Message 分為三個(gè)組成部分：
· 頭（header）是個(gè)標(biāo)準(zhǔn)字段集，客戶機(jī)和供應(yīng)商都用它來標(biāo)識(shí)和路由消息。
· 屬性（property）支持把可選頭字段添加到消息。如果您的應(yīng)用程序需要不使用標(biāo)準(zhǔn)頭字段對(duì)消息編目和分類，您就可以添加一個(gè)屬性到消息以實(shí)現(xiàn)這個(gè)編目和分類。提供 set<Type>Property(...) 和 get<Type>Property(...) 方法以設(shè)置和獲取各種 Java 類型的屬性，包括 Object。JMS 定義了一個(gè)供應(yīng)商選擇提供的標(biāo)準(zhǔn)屬性集。
· 消息的主體（body）包含要發(fā)送給接收應(yīng)用程序的內(nèi)容。每個(gè)消息接口特定于它所支持的內(nèi)容類型。
JMS 為不同類型的內(nèi)容提供了它們各自的消息類型，但是所有消息都派生自 Message 接口。
· StreamMessage：包含 Java 基本數(shù)值流，用標(biāo)準(zhǔn)流操作來順序的填充和讀取。
· MapMessage：包含一組名/值對(duì)；名稱為 string 類型，而值為 Java 的基本類型。
· TextMessage：包含一個(gè) String。
· ObjectMessage：包含一個(gè) Serializable Java 對(duì)象；能使用 JDK 的集合類。
· BytesMessage：包含未解釋字節(jié)流: 編碼主體以匹配現(xiàn)存的消息格式。
· XMLMessage: 包含XML內(nèi)容。擴(kuò)展TextMessage,XMLMessage 類型的使用，使得消息過濾非常便利。
3. 消息確認(rèn)模式
非事務(wù)性會(huì)話中，應(yīng)用程序創(chuàng)建的會(huì)話有5 種確認(rèn)模式,而在事務(wù)性會(huì)話中，確認(rèn)模式被忽略。
五種確認(rèn)模式說明:
· AUTO_ACKNOWLEDGE：自動(dòng)確認(rèn)模式。一旦接收方應(yīng)用程序的方法調(diào)用從處理消息處返回，會(huì)話對(duì)象就會(huì)確認(rèn)消息的接收。
· CLIENT_ACKNOWLEDGE：客戶端確認(rèn)模式。會(huì)話對(duì)象依賴于應(yīng)用程序?qū)Ρ唤邮盏南⒄{(diào)用一個(gè)acknowledge()方法。一旦這個(gè)方法被調(diào)用，會(huì)話會(huì)確認(rèn)最后一次確認(rèn)之后所有接收到的消息。這種模式允許應(yīng)用程序以一個(gè)調(diào)用來接收，處理并確認(rèn)一批消息。注意：在管理控制臺(tái)中，如果連接工廠的Acknowledge Policy（確認(rèn)方針）屬性被設(shè)置為"Previous"（提前），但是你希望為一個(gè)給定的會(huì)話確認(rèn)所有接收到的消息，那么就用最后一條消息來調(diào)用acknowledge()方法。
· DUPS_OK_ACKNOWLEDGE：允許副本的確認(rèn)模式。一旦接收方應(yīng)用程序的方法調(diào)用從處理消息處返回，會(huì)話對(duì)象就會(huì)確認(rèn)消息的接收；而且允許重復(fù)確認(rèn)。在需要考慮資源使用時(shí)，這種模式非常有效。注意：如果你的應(yīng)用程序無法處理重復(fù)的消息的話，你應(yīng)該避免使用這種模式。如果發(fā)送消息的初始化嘗試失敗，那么重復(fù)的消息可以被重新發(fā)送。
· NO_ACKNOWLEDGE：不確認(rèn)模式。不確認(rèn)收到的消息是需要的。消息發(fā)送給一個(gè)NO_ACKNOWLEDGE 會(huì)話后，它們會(huì)被WebLogic 服務(wù)器立即刪除。在這種模式下，將無法重新獲得已接收的消息，而且可能導(dǎo)致下面的結(jié)果：1. 消息可能丟失；和（或者）另一種情況：2. 如果發(fā)送消息的初始化嘗試失敗，會(huì)出現(xiàn)重復(fù)消息被發(fā)送的情況。
· MULTICAST_NO_ACKNOWLEDGE：IP組播下的不確認(rèn)模式，同樣無需確認(rèn)。發(fā)送給一個(gè)MULTICAST_NO_ACKNOWLEDGE會(huì)話的消息， 會(huì)共享之前所述的NO_ACKNOWLEDGE 確認(rèn)模式一樣的特征。這種模式支持希望通過IP 組播方式進(jìn)行消息通信的應(yīng)用程序，而且無需依賴會(huì)話確認(rèn)提供的服務(wù)質(zhì)量。注意：如果你的應(yīng)用程序無法處理消息的丟失或者重復(fù)，那么你應(yīng)該避免使用這種模式。如果發(fā)送消息的初始化嘗試失敗的話，重復(fù)的消息可能會(huì)被再次發(fā)送。
注：在上表的5 種確認(rèn)模式中，AUTO_ACKNOWLEDGE ，DUPS_OK_ACKNOWLEDGE 和
CLIENT_ACKNOWLEDGE 是JMS 規(guī)范定義的，NO_ACKNOWLEDGE 和MULTICAST_NO_ACKNOWLEDGE是WebLogic JMS 提供的。
三. 配置JMS
1. 創(chuàng)建連接工廠
(1) 啟動(dòng)WebLogic Server8.1,登錄控制臺(tái),選中JMS Connection Factories節(jié)點(diǎn),點(diǎn)擊右邊的" Configure a new JMS Connection Factory...";
(2) 填寫連接工廠的名稱SendJMSFactory和JNDI名稱SendJMSFactory,點(diǎn)擊"Create";
(3) 勾上"myserver",將SendJMSFactory應(yīng)用到myserver;
2. 定義后備存儲(chǔ)
(1) 選中JMS Stores節(jié)點(diǎn),點(diǎn)擊右邊的" Configure a new JMS Connection Factory...";
(2) 填寫文件后備存儲(chǔ)的名稱SendFileStore和目錄Directionary E:\BEA\user_projects\domains\mydomain\sendfilestore,點(diǎn)擊"Create".
3. 創(chuàng)建JMS服務(wù)器
(1) 選中JMS Servers節(jié)點(diǎn),點(diǎn)擊右邊的" Configure a new JMSServer...";
(2) 填寫JMS服務(wù)器的名稱SendJMSServer和Paging Store設(shè)為" SendFileStore",點(diǎn)擊"Create";
(3) Target選中"myserver",將SendJMSServer應(yīng)用到myserver.
4. 創(chuàng)建消息隊(duì)列
(1) 展開"SendJMSServer"節(jié)點(diǎn),點(diǎn)擊" Configure a new JMS Queue...";
(2) 填寫消息隊(duì)列的名稱SendJMSQueue和JNDI名稱SendJMSQueue,點(diǎn)擊"Create";
四. JMS應(yīng)用程序
一個(gè) JMS 應(yīng)用程序由下列元素組成：
· JMS 客戶機(jī)。 用 JMS API 發(fā)送和接收消息的 Java 程序。
· 非 JMS（Non-JMS）客戶機(jī)。 認(rèn)識(shí)到這一點(diǎn)很重要 - 舊的程序經(jīng)常成為整個(gè) JMS 應(yīng)用程序的一部分，而且它們的包含應(yīng)該在設(shè)計(jì)時(shí)預(yù)先考慮。
· 消息。 在 JMS 和非 JMS 客戶機(jī)之間交換的消息的格式和內(nèi)容是 JMS 應(yīng)用程序設(shè)計(jì)所必須考慮的部分。
· JMS 供應(yīng)商。供應(yīng)商必須提供特定于其 MOM 產(chǎn)品的具體的實(shí)現(xiàn)。
· 受管對(duì)象。 消息傳遞系統(tǒng)供應(yīng)商的管理員創(chuàng)建了一個(gè)對(duì)象，它獨(dú)立于供應(yīng)商專有的技術(shù)。包括連接工廠ConnectionFactory和目的Destination。
一種典型的 JMS 程序需要經(jīng)過下列步驟才能開始消息產(chǎn)生和使用:
· 通過 JNDI 查找 ConnectionFactory。
· 通過 JNDI 查找一個(gè)或多個(gè) Destination。
· 用 ConnectionFactory 創(chuàng)建一個(gè) Connection。
· 用 Connection 創(chuàng)建一個(gè)或多個(gè) Session。
· 用 Session 和 Destination 創(chuàng)建所需的 MessageProducer 和 MessageConsumer。
· 啟動(dòng) Connection。
下面利用上面配置的JMS資源演示點(diǎn)對(duì)點(diǎn)消息發(fā)送和接收的過程。
五. 設(shè)計(jì)消息發(fā)送端
1. 使用的JMS資源
服務(wù)器URL: t3://localhost:80
連接工廠: SendJMSFactory
隊(duì)列: SendJMSQueue
2. 設(shè)計(jì)步驟
· 初始化JNDI Tree
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
return new InitialContext(env);
· lookup ConnectionFactory
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
· lookup Destination
queue = (Queue) ctx.lookup(queueName);
· 用 ConnectionFactory 創(chuàng)建Connection
qcon = qconFactory.createQueueConnection();
· 用 Connection 創(chuàng)建一個(gè)Session
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
· 用 Session 和 Destination 創(chuàng)建MessageProducer
qsender = qsession.createSender(queue);
· 啟動(dòng) Connection。
qcon.start();
· 發(fā)送消息
msg = qsession.createTextMessage();
msg.setText(message);
qsender.send(msg);
3. 源代碼
package jmssample;
import java.util.Hashtable;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
/** This example shows how to establish a connection
* and send messages to the JMS queue. The classes in this
* package operate on the same JMS queue. Run the classes together to
* witness messages being sent and received, and to browse the queue
* for messages. The class is used to send messages to the queue.
*
* @author Copyright (c) 1999-2003 by BEA Systems, Inc. All Rights Reserved.
*/
public class QueueSend
{
// Defines the JNDI context factory.
public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";
// Defines the JNDI provider url.
public final static String PROVIDER_URL=" t3://localhost:80";
// Defines the JMS connection factory for the queue.
public final static String JMS_FACTORY="SendJMSFactory";
// Defines the queue.
public final static String QUEUE="SendJMSQueue";
private QueueConnectionFactory qconFactory;
private QueueConnection qcon;
private QueueSession qsession;
private QueueSender qsender;
private Queue queue;
private TextMessage msg;
/**
* Creates all the necessary objects for sending
* messages to a JMS queue.
*
* @param ctx JNDI initial context
* @param queueName name of queue
* @exception NamingException if operation cannot be performed
* @exception JMSException if JMS fails to initialize due to internal error
*/
public void init(Context ctx, String queueName)
throws NamingException, JMSException
{
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queue = (Queue) ctx.lookup(queueName);
qsender = qsession.createSender(queue);
msg = qsession.createTextMessage();
qcon.start();
}
/**
* Sends a message to a JMS queue.
*
* @param message message to be sent
* @exception JMSException if JMS fails to send message due to internal error
*/
public void send(String message) throws JMSException {
msg.setText(message);
qsender.send(msg);
}
/**
* Closes JMS objects.
* @exception JMSException if JMS fails to close objects due to internal error
*/
public void close() throws JMSException {
qsender.close();
qsession.close();
qcon.close();
}
/** main() method.
*
* @param args WebLogic Server URL
* @exception Exception if operation fails
*/
public static void main(String[] args) throws Exception {
InitialContext ic = getInitialContext();
QueueSend qs = new QueueSend();
qs.init(ic, QUEUE);
readAndSend(qs);
qs.close();
}
private static void readAndSend(QueueSend qs)
throws IOException, JMSException
{
BufferedReader msgStream = new BufferedReader(new InputStreamReader(System.in));
String line=null;
boolean quitNow = false;
do {
System.out.print("Enter message (\"quit\" to quit): ");
line = msgStream.readLine();
if (line != null && line.trim().length() != 0) {
qs.send(line);
System.out.println("JMS Message Sent: "+line+"\n");
quitNow = line.equalsIgnoreCase("quit");
}
} while (! quitNow);
}
private static InitialContext getInitialContext()
throws NamingException
{
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
return new InitialContext(env);
}
}
六. 設(shè)計(jì)消息接收端
1. 使用的JMS資源
服務(wù)器URL: t3://localhost:80
連接工廠: SendJMSFactory
隊(duì)列: SendJMSQueue
2. 設(shè)計(jì)步驟
· 初始化JNDI Tree
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
return new InitialContext(env);
· lookup ConnectionFactory
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
· lookup Destination
queue = (Queue) ctx.lookup(queueName);
· 用 ConnectionFactory 創(chuàng)建Connection
qcon = qconFactory.createQueueConnection();
· 用 Connection 創(chuàng)建一個(gè)Session
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
· 用 Session 和 Destination 創(chuàng)建MessageConsumer
qreceiver = qsession.createReceiver(queue);
· 設(shè)置監(jiān)聽
qreceiver.setMessageListener(this);
· 啟動(dòng) Connection
qcon.start();
3. 源代碼
package jmssample;
import java.util.Hashtable;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.Hashtable;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
/**
* This example shows how to establish a connection to
* and receive messages from a JMS queue. The classes in this
* package operate on the same JMS queue. Run the classes together to
* witness messages being sent and received, and to browse the queue
* for messages. This class is used to receive and remove messages
* from the queue.
*
* @author Copyright (c) 1999-2003 by BEA Systems, Inc. All Rights Reserved.
*/
public class QueueReceive implements MessageListener
{
// Defines the JNDI context factory.
public final static String JNDI_FACTORY="weblogic.jndi.WLInitialContextFactory";
// Defines the JNDI provider url.
public final static String PROVIDER_URL=" t3://localhost:80";
// Defines the JMS connection factory for the queue.
public final static String JMS_FACTORY="SendJMSFactory";
// Defines the queue.
public final static String QUEUE="SendJMSQueue";
private QueueConnectionFactory qconFactory;
private QueueConnection qcon;
private QueueSession qsession;
private QueueReceiver qreceiver;
private Queue queue;
private boolean quit = false;
/**
* Message listener interface.
* @param msg message
*/
public void onMessage(Message msg)
{
try {
String msgText;
if (msg instanceof TextMessage) {
msgText = ((TextMessage)msg).getText();
} else {
msgText = msg.toString();
}
System.out.println("Message Received: "+ msgText );
if (msgText.equalsIgnoreCase("quit")) {
synchronized(this) {
quit = true;
this.notifyAll(); // Notify main thread to quit
}
}
} catch (JMSException jmse) {
jmse.printStackTrace();
}
}
/**
* Creates all the necessary objects for receiving
* messages from a JMS queue.
*
* @param ctx JNDI initial context
* @param queueName name of queue
* @exception NamingException if operation cannot be performed
* @exception JMSException if JMS fails to initialize due to internal error
*/
public void init(Context ctx, String queueName)
throws NamingException, JMSException
{
qconFactory = (QueueConnectionFactory) ctx.lookup(JMS_FACTORY);
qcon = qconFactory.createQueueConnection();
qsession = qcon.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
queue = (Queue) ctx.lookup(queueName);
qreceiver = qsession.createReceiver(queue);
qreceiver.setMessageListener(this);
qcon.start();
}
/**
* Closes JMS objects.
* @exception JMSException if JMS fails to close objects due to internal error
*/
public void close()throws JMSException
{
qreceiver.close();
qsession.close();
qcon.close();
}
/**
* main() method.
*
* @param args WebLogic Server URL
* @exception Exception if execution fails
*/
public static void main(String[] args) throws Exception {
InitialContext ic = getInitialContext();
QueueReceive qr = new QueueReceive();
qr.init(ic, QUEUE);
System.out.println("JMS Ready To Receive Messages (To quit, send a \"quit\" message).");
// Wait until a "quit" message has been received.
synchronized(qr) {
while (! qr.quit) {
try {
qr.wait();
} catch (InterruptedException ie) {}
}
}
qr.close();
}
private static InitialContext getInitialContext()
throws NamingException
{
Hashtable env = new Hashtable();
env.put(Context.INITIAL_CONTEXT_FACTORY, JNDI_FACTORY);
env.put(Context.PROVIDER_URL, PROVIDER_URL);
return new InitialContext(env);
}
}
七. 測(cè)試消息發(fā)送和接收
1. 設(shè)置WebLogic Classpath;
2. 轉(zhuǎn)到發(fā)送接收程序目錄編譯文件;
3. 執(zhí)行接受程序;
4. 打開另一窗口,執(zhí)行發(fā)送程序;
5. 輸入發(fā)送消息"quit",接收程序結(jié)束.
總結(jié)
本文先簡(jiǎn)要介紹了JMS的一些基本概念,繼而引入了WebLogic JMS Server的體系結(jié)構(gòu)和相關(guān)特性。在此基礎(chǔ)之上,圖文并茂地講述了JMS在WebLogic Server 8.1上的配置。最后在解剖JMS應(yīng)用程序框架的同時(shí),以點(diǎn)對(duì)點(diǎn)為例演示了JMS的發(fā)送接收消息流程。
|