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

作者：orangelizq
 

2.3 消息中間件的傳遞模式

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

1.        點對點模型（PTP）

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

2.        發布-訂閱模型（Pub/Sub）

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

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

多個應用程序可以就一個主題發布和訂閱消息，而應用程序對其他人仍然是匿名的。MOM 起著代理（broker）的作用，將一個主題已發表的消息路由給該主題的所有訂閱者。

2.4 消息中間件產品與JMS

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

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

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

1．IBM MQSeries

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

2．WebLogic

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

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

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

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

4)        支持XML消息，動態創建持久隊列和主題。

3．SonicMQ

SonicMQ是Progress公司實現的JMS產品。除了提供基本的消息驅動服務之外，SonicMQ也提供了很多額外的企業級應用開發工具包，它具有以下一些基本特征：

1)        提供JMS規范的完全實現，支持點到點消息模式和發布/訂閱消息模式；

2)        支持層次安全管理；

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

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

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

4．Active MQ

Active MQ是一個基于Apcache 2.0 licenced發布，開放源碼的JMS產品。其特點為：

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

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

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

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

5．OpenJMS

OpenJMS是一個開源的JMS規范的實現，它包含以下幾個特征：

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

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

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

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

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

三、消息中間件應用之JMS

3.1 JMS簡介

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

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

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

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

統一模式的好處是：

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

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

3)        為JMS提供者提供了優化其實現的機會。

3.2 JMS體系結構

3.2.1 JMS接口描述

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

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

ConnectionFactory：連接工廠，JMS 用它創建連接

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

Destination：消息的目的地

Session：一個發送或接收消息的線程

MessageProducer： 由Session 對象創建的用來發送消息的對象

MessageConsumer： 由Session 對象創建的用來接收消息的對象

3.2.2 JMS消息模型

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

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

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

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

3.3 JMS編程實踐

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

1)        用JNDI 得到ConnectionFactory對象；

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

3)        用ConnectionFactory創建Connection 對象；

4)        用Connection對象創建一個或多個JMS Session；

5)        用Session 和Destination 創建MessageProducer和MessageConsumer；

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

3.3.1 消息生產者編程

消息生產者程序如下：^[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;

        }

    }

}

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

四、消息中間件總結

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

參考文獻

[1] 李華琰、郭英奎 著：Java 中間件開發技術 . 中國水利水電出版社，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的數據匯集系統的研究與實現.  2006. 

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