消息中间件提供企业数据的异步传输，通过消息中间件，一些原本互相孤立的业务组件可以组合成一个可靠的、灵活的系统。

消息中间件大致分为两类：Point-to-Point(PTP)和Publish-Subscribe(Pub/Sub),Apusic支持这两种模型。

PTP是点对点传输消息，建立在消息队列的基础上，每个客户端对应一个消息队列，客户端发送消息到对方的消息队列中，从自己的消息队列读取消息。

Pub/Sub是将消息定位到某个层次结构栏目的节点上，Pub/Sub通常是匿名的并能够动态发布消息，Pub/Sub必须保证某个节点的所有发布者(Publisher)发布的信息准确无误地发送到这个节点的所有消息订阅者(Subscriber)。

JMS 支持两种消息类型PTP 和Pub/Sub，分别称作：PTP Domain 和Pub/Sub Domain，这两种接口都继承统一的JMS Parent 接口，JMS 主要接口如下所示：

以下是对这些接口的简单描述：

ConnectionFactory ：连接工厂，JMS 用它创建连接

Connection ：JMS 客户端到JMS Provider 的连接

Destination ：消息的目的地

Session： 一个发送或接收消息的线程

MessageProducer： 由Session 对象创建的用来发送消息的对象

MessageConsumer： 由Session 对象创建的用来接收消息的对象

JDBC： JMS 客户端可以使用JDBC 接口，可以将JDBC 和JMS包含在一个事务里。这种包含可以在EJB里，也可以直接调用JTA(Java Transaction API)接口实现。

JavaBeans： JavaBeans可以用JMS Session 发送接收消息。

EJB： EJB2.1 规范中定义了新的Message-Driven Beans 组件模型。

JTA(Java Transaction API)： JMS 客户端可以用JTA 启动事务。JMS Provider可以选择是否支持分布式事务。

JTS(Java Transaction Service)： JMS 可以和JTS一起组成一个分布式事务，如将发送接收消息和更新数据库包含在一个事务里。

JNDI：JMS客户端通过JNDI 调用JMS 中的对象。

JMS 消息由以下几部分组成：消息头，属性，消息体。

消息头(Header) -消息头包含消息的识别信息和路由信息，消息头包含一些标准的属性如：JMSDestination,JMSMessageID等。

属性(Properties) -除了消息头中定义好的标准属性外，JMS提供一种机制增加新属性到消息头中，这种新属性包含以下几种：

标准的JMS 消息头包含以下属性：

JMSDestination --消息发送的目的地。

JMSDeliveryMode --传递模式， 有两种模式：PERSISTENT和NON_PERSISTENT，PERSISTENT表示该消息一定要被送到目的地，否则会导致应用错误。NON_PERSISTENT表示偶然丢失该消息是被允许的，这两种模式使开发者可以在消息传递的可靠性和吞吐量之间找到平衡点。

JMSMessageID 唯一识别每个消息的标识，由JMS Provider 产生。

JMSTimestamp 一个消息被提交给JMS Provider 到消息被发出的时间。

JMSCorrelationID 用来连接到另外一个消息，典型的应用是在回复消息中连接到原消息。

JMSReplyTo 提供本消息回复消息的目的地址。

JMSRedelivered如果一个客户端收到一个设置了JMSRedelivered属性的消息，则表示可能该客户端曾经在早些时候收到过该消息，但并没有签收(acknowledged)。

JMSType 消息类型的识别符。

JMSExpiration 消息过期时间，等于QueueSender 的send方法中的timeToLive值或TopicPublisher 的publish 方法中的timeToLive值加上发送时刻的GMT时间值。如果timeToLive值等于零，则JMSExpiration被设为零，表示该消息永不过期。如果发送后，在消息过期时间之后消息还没有被发送到目的地，则该消息被清除。

JMSPriority 消息优先级，从0-9 十个级别，0-4 是普通消息，5-9 是加急消息。JMS 不要求JMS Provider严格按照这十个优先级发送消息，但必须保证加急消息要先于普通消息到达。

消息体(Body) - JMS API定义了5种消息体格式，也叫消息类型，你可以使用不同形式发送接收数据并可以兼容现有的消息格式，下面描述这5种类型：

下例演示创建并发送一个TextMessage到一个队列：

TextMessage message = queueSession.createTextMessage();
message.setText(msg_text); // msg_text is a String
queueSender.send(message);

下例演示接收消息并转换为合适的消息类型：

Message m = queueReceiver.receive();
if (m instanceof TextMessage) {
	TextMessage message = (TextMessage) m;
	System.out.println("Reading message: " + message.getText());
} else {
	// Handle error
}

消息的同步接收

同步接收是指客户端主动去接收消息，JMS 客户端可以采用MessageConsumer的receive方法去接收下一个消息。

消息的异步接收

异步接收是指当消息到达时，主动通知客户端。JMS客户端可以通过注册一个实现MessageListener接口的对象到MessageConsumer，这样，每当消息到达时，JMS Provider 会调用MessageListener中的onMessage 方法。

PTP(Point-to-Point)模型是基于队列的，发送方发消息到队列，接收方从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提供工具管理队列的创建、删除。JMS PTP模型定义了客户端如何向队列发送消息，从队列接收消息，浏览队列中的消息。

下面描述JMS PTP 模型中的主要概念和对象：

JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作主题(topic)。

主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscribe)从主题订阅消息。主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。

下面描述JMS Pub/Sub 模型中的主要概念和对象：

广义上说，一个JMS 应用是几个JMS 客户端交换消息，开发JMS 客户端应用由以下几步构成：

1. 用JNDI 得到ConnectionFactory 对象；

2. 用JNDI 得到目标队列或主题对象，即Destination 对象；

3. 用ConnectionFactory 创建Connection 对象；

4. 用Connection 对象创建一个或多个JMS Session；

5. 用Session 和Destination 创建MessageProducer和MessageConsumer；

6. 通知Connection 开始传递消息。

PTP 模型主要包含消息的发送和接收，下面我们分别举例说明：　

发送消息

第一，启动Apusic应用服务器。

第二，打开管理工具：在浏览器中输入http://localhost:6888，进到Web管理，缺省管理员为admin，密码也是admin。

第三，配置队列连接创建器：在管理工具中点击“管理控制台”-->“消息服务”，点击“连接创建器”，如果存在JNDI名为“jms/QueueConnectionFactory”的队列连接创建器，则无须创建，否则点击“创建”，添加必要信息，“连接创建器名称”为“QueueConnectionFactory”，“队列连接创建器JNDI名”为“jms/QueueConnectionFactory”，“连接是否可以匿名访问”为“是”，其它项为缺省设置，点击“保存”。如下：

连接创建器

第四，配置队列：在管理工具中点击“管理控制台”-->“消息服务”，点击“队列（queue）”，如果存在JNDI名为“jms/QueueConnectionFactory”的队列，则无须创建，否则点击“创建”，添加必要信息，“队列名称”为“testQueue”，“队列JNDI名”为“testQueue”，其它项为缺省设置，点击“保存”。如下：

配置队列

第五，编写发送消息客户端代码：Send.java

import javax.jms.*;
import javax.naming.*;
import java.io.*;
import java.util.*;
import java.rmi.RemoteException;
/**
 * Title:        JMS
 * Description:  JMS Test
 * Copyright:    Copyright (c) 2003
 * Company:      Apusic
 * @author Michael
 * @version 1.0
 */

public class Send{

   String			queueName = "myQueue";
   QueueConnectionFactory	queueConnectionFactory = null;
   Queue			queue = null;
   QueueConnection		queueConnection = null;
   QueueSession		queueSession = null;
   QueueSender		queueSender = null;
   TextMessage      		message = null;

   public static void main(String[] args) throws Exception {

	InitialContext ic = getInitialContext();
	Send sender = new Send();
	sender.init(ic) ;
	sender.sendMessage();
	sender.close();

   }

   public void init(InitialContext ctx) throws Exception{

        queueConnectionFactory = (QueueConnectionFactory)ctx.lookup("jms/QueueConnectionFactory");
        queueConnection = queueConnectionFactory.createQueueConnection();
        queue = (Queue) ctx.lookup(queueName);
   }

   public void sendMessage() throws  JMSException,RemoteException{

        queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
        queueSender = queueSession.createSender(queue);
        queueSender.setDeliveryMode(DeliveryMode.PERSISTENT);
        message = queueSession.createTextMessage();
        message.setText("The Message from myQueue");
        queueSender.send(message);
   }

   public void close() throws JMSException{

        if(queueConnection!=null)
        	queueConnection.close();
   }

   private static InitialContext getInitialContext() throws NamingException{

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY,"com.apusic.naming.jndi.CNContextFactory");
        env.put(Context.PROVIDER_URL, "iiop://localhost:6888");
        return (new InitialContext(env));

   }
}

接收消息

如果管理工具中没有JNDI名为“jms/QueueConnectionFactory”的队列连接创建器和“myQueue”的队列，则依照上面一到四步进行设置，否则可直接编写接收消息客户端代码：Receive.java

import javax.jms.*;
import javax.naming.*;
import java.io.*;
import java.util.*;
import java.rmi.RemoteException;
/**
 * Title:        JMS
 * Description:  JMS Test
 * Copyright:    Copyright (c) 2003
 * Company:      Apusic
 * @author Michael
 * @version 1.0
 */

public class Receive{

   String			queueName = "myQueue";
   QueueConnectionFactory	queueConnectionFactory = null;
   Queue			queue = null;
   QueueConnection		queueConnection = null;
   QueueSession		queueSession = null;
   QueueReceiver		queueReceiver = null;
   TextMessage      		message = null;

   public static void main(String[] args) throws Exception {

        InitialContext ic = getInitialContext();
        Receive receiver = new Receive();
        receiver.init(ic) ;
        receiver.TBreceiveMessage();//你可以在此处调用YBreceiveMessage
        receiver.close();
   }

   public void init(InitialContext ctx) throws Exception{

        queueConnectionFactory = (QueueConnectionFactory)ctx.lookup("jms/QueueConnectionFactory");
        queueConnection = queueConnectionFactory.createQueueConnection();
        queue = (Queue) ctx.lookup(queueName);
   }

   public void TBreceiveMessage() throws NamingException, JMSException,RemoteException{

        queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
  	 	queueReceiver = queueSession.createReceiver(queue);
        queueConnection.start();
        for (;;) {
            message = (TextMessage) queueReceiver.receive();
            System.out.println("Reading message: " + message.getText());
            if (message.getText().equals("quit"))
                break;
        }
   }

   public void YBreceiveMessage() throws NamingException, JMSException,RemoteException,IOException{

        queueSession = queueConnection.createQueueSession(false,Session.AUTO_ACKNOWLEDGE);
        queueReceiver = queueSession.createReceiver(queue);
        //register my textListener which comes from MessageListener
        TextMessageListener textListener = new TextMessageListener();
        queueReceiver.setMessageListener(textListener);
        queueConnection.start();

        System.out.println("To end program, enter Q or q, then ");
        InputStreamReader reader = new InputStreamReader(System.in);
        char answer = '\0';

        while (!((answer == 'q') || (answer == 'Q')))
        	answer = (char)reader.read();
   }

   public void close() throws JMSException{

	 	if(queueReceiver!=null)
	 		queueReceiver.close();
	 	if(queueSession!=null)
	 		queueSession.close();
        if(queueConnection!=null)
        	queueConnection.close();
   }

   private static InitialContext getInitialContext() throws NamingException{

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY,
               "com.apusic.naming.jndi.CNContextFactory");
        env.put(Context.PROVIDER_URL, "iiop://localhost:6888");
        return (new InitialContext(env));

   }
}

其中异步接收时使用的TextMessageListener代码如下：

import javax.jms.MessageListener;
import javax.jms.Message;
import javax.jms.TextMessage;
import javax.jms.JMSException;

/**
 * Title:        JMS
 * Description:  JMS Test
 * Copyright:    Copyright (c) 2003
 * Company:      Apusic
 * @author Michael
 * @version 1.0
 */

public class TextMessageListener implements MessageListener {

    public TextMessageListener() {}

    public void onMessage(Message m) {

        TextMessage msg = (TextMessage) m;
        try {
	       System.out.println("Async reading message: " + msg.getText() +
			       " (priority=" + msg.getJMSPriority() + ")");
	    } catch (JMSException e) {
	       System.out.println("Exception in onMessage(): " + e.toString());
	    }
    }
}

最后，运行程序：在Windows下打开两个DOS窗口，首先运行Receive 然后运行Send。

如果是异步接收消息，在运行Receive 的窗口中可以看到输出：

To end program, enter Q or q, then
Async reading message: The Second Message from testQueue (priority=4)

如果是同步接收消息，在运行Receive 的窗口中可以看到输出：

Reading message: The Message from testQueue

Pub/Sub 模型主要包含消息的发布和订阅，下面我们分别举例说明：

发布消息

如果管理工具中没有JNDI名为“jms/TopicConnectionFactory”的连接创建器和“myTopic”的队列，则依照上面一到四步进行设置，否则可直接编写发布消息客户端代码：Published.java

import javax.jms.*;
import javax.naming.*;
import java.io.*;
import java.util.*;
import java.rmi.RemoteException;
/**
 * Title:        JMS
 * Description:  JMS Test
 * Copyright:    Copyright (c) 2003
 * Company:      Apusic
 * @author Michael
 * @version 1.0
 */

public class Published{

    String                       topicName = "myTopic";
    TopicConnectionFactory       topicConnectionFactory = null;
    Topic                        topic = null;
    TopicConnection              topicConnection = null;
    TopicSession                 topicSession = null;
    TopicPublisher               topicPublisher = null;
    String                       msgText = null;
    TextMessage                  message = null;

    public static void main(String[] args) throws Exception {

        InitialContext ic = getInitialContext();
        Published publisher = new Published();
        publisher.init(ic) ;
        publisher.publish();
        publisher.close();
    }

    public void init(InitialContext ctx) throws Exception{

        topicConnectionFactory = (TopicConnectionFactory)ctx.lookup("jms/TopicConnectionFactory");
        topicConnection = topicConnectionFactory.createTopicConnection();
        topic = (Topic) ctx.lookup(topicName);
    }

    public void publish() throws NamingException, JMSException,RemoteException{

        topicSession = topicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
        topicPublisher = topicSession.createPublisher(topic);
        message = topicSession.createTextMessage();
        msgText = "This is the published message";
        message.setText(msgText);
        topicPublisher.publish(message);
    }

    public void close() throws JMSException{

        if(topicConnection!=null)
	topicConnection.close();
    }

    private static InitialContext getInitialContext() throws NamingException{

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY,"com.apusic.naming.jndi.CNContextFactory");
        env.put(Context.PROVIDER_URL, "iiop://localhost:6888");
        return (new InitialContext(env));

    }

}

订阅消息

import javax.jms.*;
import javax.naming.*;
import java.io.*;
import java.util.*;
import java.rmi.RemoteException;
/**
 * Title:        JMS
 * Description:  JMS Test
 * Copyright:    Copyright (c) 2003
 * Company:      Apusic
 * @author Michael
 * @version 1.0
 */

public class Subscriber{

    String                       topicName = "myTopic";
    TopicConnectionFactory       topicConnectionFactory = null;
    TopicConnection              topicConnection = null;
    Topic                        topic = null;
    TopicSession                 topicSession = null;
    TopicSubscriber              topicSubscriber = null;
    TextMessage                  message = null;

    String id = "durable";

    public static void main(String[] args) throws Exception {

        InitialContext ic = getInitialContext();
        Subscriber subscriber = new Subscriber();
        subscriber.init(ic) ;
        subscriber.subscribe();
        subscriber.close();
    }

    public void init(InitialContext ctx) throws Exception{

        topicConnectionFactory = (TopicConnectionFactory)ctx.lookup("jms/TopicConnectionFactory");
        topicConnection = topicConnectionFactory.createTopicConnection();
        topicConnection.setClientID(id) ;
        topic = (Topic) ctx.lookup(topicName);
    }

    public void subscribe() throws NamingException, JMSException,RemoteException{

        topicSession = topicConnection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);
        topicSubscriber = topicSession.createDurableSubscriber(topic,id);
        topicConnection.start();
        message = (TextMessage) topicSubscriber.receive();
        System.out.println("SUBSCRIBER THREAD: Reading message: " + message.getText());

    }

    public void close() throws JMSException{

        if(topicConnection!=null)
	topicConnection.close();
    }

    private static InitialContext getInitialContext() throws NamingException{

        Hashtable env = new Hashtable();
        env.put(Context.INITIAL_CONTEXT_FACTORY,"com.apusic.naming.jndi.CNContextFactory");
        env.put(Context.PROVIDER_URL, "iiop://localhost:6888");
        return (new InitialContext(env));

    }

}

最后，运行程序：在Windows下打开两个DOS窗口，首先运行Subscriber，然后运行Published，在运行Subscriber的窗口中可以看到输出：

SUBSCRIBER THREAD: Reading message: This is the published message

利用Apusic JMS配置文件（jms.xml），你可以对队列和主题设定访问权限，可以详细规定Apusic中的用户和组，可以对队列和主题进行何种操作。

对于队列，可以设定发送(send)、接收(receive)和浏览(browse)权限；对于主题，可以设定发布(publish)、订阅(subscribe)、持久订阅(subscribe-durable)和取消持久订阅(unsubscribe)权限。

客户端用QueueConnectionFactory 的createQueueConnection(java.lang.String userName,java.lang.String password)方法创建队列连接时输入用户身份,或者用createTopicConnection(java.lang.String userName, java.lang.String password)方法创建主题连接时输入用户身份。

下面的例子授权用户larry 可以对队列testQueue 进行接收和浏览：

首先创建testQueue队列，如果Apusic不存在testQueue队列，则通过管理工具创建，方法同上。

建立安全角色(security-role)rbt，将该角色映射到用户larry，使larry拥有该角色包含的所有权限。如果Apuisc应用服务器不存在larry用户，在管理工具中“用户管理”-->“用户/组”中创建larry用户。然后可参考apusic.jar中jms.dtd的定义在jms.xml中加入如下片段：

<security-role>
    <role-name>rbt</role-name>
    <principal>larry</principal>
</security-role>

设置安全角色rbt的权限：

<destination-permission>
    <role-name>rbt</role-name>
    <destination-method>
        <queue-name>testQueue</queue-name>
        <method-name>receive</method-name>
        <method-name>browse</method-name>
    </destination-method>
</destination-permission>