消息中間件原理及JMS簡介之二

作者：orangelizq
 

2.3 消息中間件的傳遞模式

消息中間件一般有兩種傳遞模型：點對點模型（PTP）和發(fā)布-訂閱模型（Pub/Sub）^[2]。

1.        點對點模型（PTP）

點對點模型用于消息生產(chǎn)者和消息消費者之間點到點的通信。消息生產(chǎn)者將消息發(fā)動到由某個名字標識的特定消費者。這個名字實際上對應于消息服務中的一個隊列（Queue），在消息傳動給消費者之前它被存儲在這個隊列中。隊列可以是持久的，以保證在消息服務出現(xiàn)故障時仍然能夠傳遞消息。

2.        發(fā)布-訂閱模型（Pub/Sub）

發(fā)布-訂閱模型用稱為主題（topic）的內(nèi)容分層結構代替了PTP模型中的惟一目的地，發(fā)送應用程序發(fā)布自己的消息，指出消息描述的是有關分層結構中的一個主題的信息。希望接收這些消息的應用程序訂閱了這個主題。訂閱包含子主題的分層結構中的主題的訂閱者可以接收該主題和其子主題發(fā)表的所有消息。

下圖展示了發(fā)布和訂閱模型：^[2]

多個應用程序可以就一個主題發(fā)布和訂閱消息，而應用程序?qū)ζ渌巳匀皇悄涿摹?/span>MOM 起著代理（broker）的作用，將一個主題已發(fā)表的消息路由給該主題的所有訂閱者。

2.4 消息中間件產(chǎn)品與JMS

從上個世紀90年代初，隨著不同廠商消息中間件大量上市，消息中間件技術得到了長足的發(fā)展。目前，IBM和BEA的中間件產(chǎn)品在銀行、證券、電信等高端行業(yè)，以及IT等行業(yè)中得到廣泛應用。IBM憑借其在1999年推出的應用服務器WebSphere，扎根金融、證券等行業(yè)，在超大型以及系統(tǒng)整合型應用方面優(yōu)勢突出；BEA則是專門從事中間件開發(fā)的公司，它的應用服務器WebLogic在美國市場占有率超過60％，在國內(nèi)電信及證券行業(yè)占據(jù)主要地位；Sun、Oracle、Sybase和Borland等廠商也都有自己的應用服務器；近年來，以金蝶、東方通等公司為代表的國產(chǎn)中間件產(chǎn)品也發(fā)展迅速。^[3]

由于沒有統(tǒng)一的規(guī)范和標準，基于消息中間件的應用不可移植，不同的消息中間件也不能互操作，這大大阻礙了消息中間件的發(fā)展。                   Java Message Service(JMS, Java消息服務)是SUN及其伙伴公司提出的旨在統(tǒng)一各種消息中間件系統(tǒng)接口的規(guī)范。它定義了一套通用的接口和相關語義，提供了諸如持久、驗證和事務的消息服務，它最主要的目的是允許Java應用程序訪問現(xiàn)有的消息中間件。JMS規(guī)范沒有指定在消息節(jié)點間所使用的通訊底層協(xié)議，來保證應用開發(fā)人員不用與其細節(jié)打交道，一個特定的JMS實現(xiàn)可能提供基于TCP/IP、HTTP、UDP或者其它的協(xié)議。

目前許多廠商采用并實現(xiàn)了JMS API，現(xiàn)在，JMS產(chǎn)品能夠為企業(yè)提供一套完整的消息傳遞功能，下面是一些比較流行的JMS商業(yè)軟件和開源產(chǎn)品。

1．IBM MQSeries

IBM MQ系列產(chǎn)品提供的服務使得應用程序可以使用消息隊列進行相互交流，通過一系列基于Java的API，提供了MQSeries在Java中應用開發(fā)的方法。它支持點到點和發(fā)布/訂閱兩種消息模式，在基本消息服務的基礎上增加了結構化消息類，通過工作單元提供數(shù)據(jù)整合等內(nèi)容。

2．WebLogic

WebLogic是BEA公司實現(xiàn)的基于工業(yè)標準的J2EE應用服務器，支持大多數(shù)企業(yè)級JavaAPI，它完全兼容JMS規(guī)范，支持點到點和發(fā)布/訂閱消息模式，它具有以下一些特點：

1)        通過使用管理控制臺設置JMS配置信息；

2)        支持消息的多點廣播；

3)        支持持久消息存儲的文件和數(shù)據(jù)庫；

4)        支持XML消息，動態(tài)創(chuàng)建持久隊列和主題。

3．SonicMQ

SonicMQ是Progress公司實現(xiàn)的JMS產(chǎn)品。除了提供基本的消息驅(qū)動服務之外，SonicMQ也提供了很多額外的企業(yè)級應用開發(fā)工具包，它具有以下一些基本特征：

1)        提供JMS規(guī)范的完全實現(xiàn)，支持點到點消息模式和發(fā)布/訂閱消息模式；

2)        支持層次安全管理；

3)        確保消息在Internet上的持久發(fā)送；

4)        動態(tài)路由構架（DRA）使企業(yè)能夠通過單個消息服務器動態(tài)的交換消息；

5)        支持消息服務器的集群。

4．Active MQ

Active MQ是一個基于Apcache 2.0 licenced發(fā)布，開放源碼的JMS產(chǎn)品。其特點為：

1)        提供點到點消息模式和發(fā)布/訂閱消息模式；

2)        支持JBoss、Geronimo等開源應用服務器，支持Spring框架的消息驅(qū)動；

3)        新增了一個P2P傳輸層，可以用于創(chuàng)建可靠的P2P JMS網(wǎng)絡連接；

4)        擁有消息持久化、事務、集群支持等JMS基礎設施服務。

5．OpenJMS

OpenJMS是一個開源的JMS規(guī)范的實現(xiàn)，它包含以下幾個特征：

1)        它支持點到點模型和發(fā)布/訂閱模型；

2)        支持同步與異步消息發(fā)送；

3)        可視化管理界面，支持Applet；

4)        能夠與Jakarta Tomcat這樣的Servlet容器結合；

5)        支持RMI、TCP、HTTP與SSL協(xié)議。

三、消息中間件應用之JMS

3.1 JMS簡介

Java Message Service 規(guī)范 1.1 聲稱：JMS 是一組接口和相關語義，它定義了 JMS 客戶如何訪問企業(yè)消息產(chǎn)品的功能。

       在 JMS 之前，每一家 MOM 廠商都用專有 API 為應用程序提供對其產(chǎn)品的訪問，通常可用于許多種語言，其中包括 Java 語言。JMS 通過 MOM 產(chǎn)品為 Java 程序提供了一個發(fā)送和接收消息的標準的、便利的方法。用 JMS 編寫的程序可以在任何實現(xiàn) JMS 標準的 MOM 上運行。

JMS 可移植性的關鍵在于：JMS API 是由 Sun 作為一組接口而提供的。提供了 JMS 功能的產(chǎn)品是通過提供一個實現(xiàn)這些接口的提供者來做到這一點的。開發(fā)人員可以通過定義一組消息和一組交換這些消息的客戶機應用程序建立 JMS 應用程序。^[4]

JMS1.0版本于1998年推出，最新的版本是2002發(fā)布的JMS 1.1規(guī)范。JMS支持消息中間件的兩種傳遞模式：點到點模式和發(fā)布-訂閱模式。在JMS 1.1以前的版本中，每一種都有自己的特定于該模式的一組客戶機接口。JMS1.1版本提供了單一的一組接口，它允許客戶機可以在兩個模式中發(fā)送和接收消息。這些“模式無關的接口”保留了每一個模式的語義和行為，是實現(xiàn) JMS 客戶機的最好選擇。

統(tǒng)一模式的好處是：

1)        使得用于客戶機的編程更簡單。

2)        隊列和主題的操作可以是同一事務的一部分。

3)        為JMS提供者提供了優(yōu)化其實現(xiàn)的機會。

3.2 JMS體系結構

3.2.1 JMS接口描述

JMS 支持兩種消息類型PTP 和Pub/Sub，分別稱作：PTP Domain 和Pub/Sub Domain，這兩種接口都繼承統(tǒng)一的JMS Parent 接口，JMS 主要接口如下所示：

以下是對這些接口的簡單描述：

ConnectionFactory：連接工廠，JMS 用它創(chuàng)建連接

Connection：JMS 客戶端到JMS Provider 的連接

Destination：消息的目的地

Session：一個發(fā)送或接收消息的線程

MessageProducer： 由Session 對象創(chuàng)建的用來發(fā)送消息的對象

MessageConsumer： 由Session 對象創(chuàng)建的用來接收消息的對象

3.2.2 JMS消息模型

JMS 消息由以下幾部分組成：消息頭，屬性，消息體。^[4]

l        消息頭（header）：JMS消息頭包含了許多字段，它們是消息發(fā)送后由JMS提供者或消息發(fā)送者產(chǎn)生，用來表示消息、設置優(yōu)先權和失效時間等等，并且為消息確定路由。

l        屬性（property）：由消息發(fā)送者產(chǎn)生，用來添加刪除消息頭以外的附加信息。

l        消息體（body）：由消息發(fā)送者產(chǎn)生，JMS中定義了5種消息體：ByteMessage、MapMessage、ObjectMessage、StreamMessage和TextMessage。
  

3.3 JMS編程實踐

廣義上說，一個JMS應用是幾個JMS 客戶端交換消息，開發(fā)JMS客戶端應用由以下幾步構成：^[2]

1)        用JNDI 得到ConnectionFactory對象；

2)        用JNDI 得到目標隊列或主題對象，即Destination對象；

3)        用ConnectionFactory創(chuàng)建Connection 對象；

4)        用Connection對象創(chuàng)建一個或多個JMS Session；

5)        用Session 和Destination 創(chuàng)建MessageProducer和MessageConsumer；

6)        通知Connection 開始傳遞消息。

3.3.1 消息生產(chǎn)者編程

消息生產(chǎn)者程序如下：^[2]

package org.jms.test;

import java.io.*;

import javax.jms.*;

import javax.naming.*;

public class Sender {

    public static void main(String[] args) {

        new Sender().send();

    }

    public void send() {

        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        try {

            //Prompt for JNDI names

            System.out.println("Enter ConnectionFactory name:");

            String factoryName = reader.readLine();

            System.out.println("Enter Destination name:");

            String destinationName = reader.readLine();

            //Look up administered objects

            InitialContext initContext = new InitialContext();

            ConnectionFactory factory =

                (ConnectionFactory) initContext.lookup(factoryName);

            Destination destination = (Destination) initContext.lookup(destinationName);

            initContext.close();

            //Create JMS objects

            Connection connection = factory.createConnection();

            Session session =

                connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            MessageProducer sender = session.createProducer(queue);

            //Send messages

            String messageText = null;

            while (true) {

                System.out.println("Enter message to send or 'quit':");

                messageText = reader.readLine();

                if ("quit".equals(messageText))

                    break;

                TextMessage message = session.createTextMessage(messageText);

                sender.send(message);

            }

            //Exit

            System.out.println("Exiting...");

            reader.close();

            connection.close();

            System.out.println("Goodbye!");

        } catch (Exception e) {

            e.printStackTrace();

            System.exit(1);

        }

    }

}

3.3.2 消息消費者編程

消息消費者程序如下：^[2]

package compute;

import java.io.*;

import javax.jms.*;

import javax.naming.*;

public class Receiver implements MessageListener {

    private boolean stop = false;

    public static void main(String[] args) {

        new Receiver().receive();

    }

    public void receive() {

        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        try {

            //Prompt for JNDI names

            System.out.println("Enter ConnectionFactory name:");

            String factoryName = reader.readLine();

            System.out.println("Enter Destination name:");

            String destinationName = reader.readLine();

            reader.close();

            //Look up administered objects

            InitialContext initContext = new InitialContext();

            ConnectionFactory factory =

                (ConnectionFactory) initContext.lookup(factoryName);

            Destination destination = (Destination) initContext.lookup(destinationName);

            initContext.close();

            //Create JMS objects

            Connection connection = factory.createConnection();

            Session session =

                connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            MessageConsumer receiver = session.createConsumer(queue);

            receiver.setMessageListener(this);

            connection.start();

            //Wait for stop

            while (!stop) {

                Thread.sleep(1000);

            }

            //Exit

            System.out.println("Exiting...");

            connection.close();

            System.out.println("Goodbye!");

        } catch (Exception e) {

            e.printStackTrace();

            System.exit(1);

        }

    }

    public void onMessage(Message message) {

        try {

            String msgText = ((TextMessage) message).getText();

            System.out.println(msgText);

            if ("stop".equals(msgText))

                stop = true;

        } catch (JMSException e) {

            e.printStackTrace();

            stop = true;

        }

    }

}

以上程序都較為簡單，基本上為自解釋的。

四、消息中間件總結

消息中間件自從產(chǎn)生以來發(fā)展迅速，在分布式聯(lián)機事務處理環(huán)境中，它擔當通訊資源管理器（CRM）的角色，為分布式應用提供實時、高效、可靠的、跨越不同操作系統(tǒng)、不同網(wǎng)絡的消息傳遞服務，同時消息中間件減少了開發(fā)跨平臺應用程序的復雜性。在要求可靠傳輸?shù)南到y(tǒng)中可以利用消息中間件作為一個通訊平臺，向應用提供可靠傳輸功能來傳遞消息和文件。

參考文獻

[1] 李華琰、郭英奎 著：Java 中間件開發(fā)技術 . 中國水利水電出版社，2005

[2] JMS在線教程 . 2004  http://www6.software.ibm.com/developerworks/cn/education/java/j-jms

[3] 基于JMS的消息中間件的研究和設計 . 姚剛. 2006

[4] 消息中間件和JMS . 2005.  http://www.huihoo.org/jfox/jfoxmq/mom_jms.html

[5] 基于JMS的數(shù)據(jù)匯集系統(tǒng)的研究與實現(xiàn).  2006. 

       http://www.cndw.com/tech/program/2006042760285.asp