Java JMS学习纪要

http://www.51testing.com/html/39/n-817539.html

Java消息服务（JMS）学习小结

 

　最近，有一个项目要用到JMS，故而稍微学习了一点。

　　消息服务是指，两个或者多个客户机能够通过发送和接受消息（以对等的方式）通信。消息是通过消息服务器有一个客户机发送到另一个客户机的“一块”数据，可以是文本的，也可是数值的，如果客户机是Java应用程序，数据还可以包括对象。其中客户机不需要同时运行。

　　使用消息服务的原因：

　　1、松散耦合但是高内聚。使用消息服务的客户机不需要实现通用接口，不需要了解对方。消息服务提供了标准接口。

　　2、不直接通信。客户机不直接对话，而通过中间媒介，消息服务扮演 缓冲区，并提供 安全控制。

　　3、保证消息传递。 JMS的提供者保持消息持久，直到客户机接受为止。

　　4、异步通信。

　　5、一对多、多对多和多对一通信。

　　JMS（Java 消息服务）是一组标准的API，能够用于访问多种消息服务器。使用JMS，可以使用一样的API访问IBM的MQSeries、JBossMQ等消息服务。

　　JMS API中有很多核心概念是映射到底层的消息服务器。其中：

　　1、受控对象（Administered Object）。它们是由管理者创建的供JMS客户机使用的对象。如 连接工厂（Connection Factory），它们用于与底层的消息服务器和目标（队列和主题）进行连接（注：不太理解，可能错了）。管理通过JNDI对他们进行管理。JMS提供了JMS客户机和实际JMS提供者之间的缓冲区。JBoss就是这些对象的管理者。

　　2、JMS提供者。实现了JMS接口的消息服务器。（如 JBossMQ消息服务器）。

　　3、JMS客户机。消息的生产者和消费者。由于是对等的通信机制，所以没有客户机和服务器的概念。JMS既可以是消息创建者又是消息接收者。

　　4、消息（Message）。JMS客户机之间传送的一条条消息。

　　传统的消息服务一般支持点对点通信和发布/订阅通信两种通信模式的一种。JMS API 两种都支持。

　　Point-To-Point

　　点对点通信模式，有一个中心队列作为发布的目标（受控对象）。一个或多个消息生产者可以发送消息到这个队列。然后被消息的消费者选取。

　　Pub/Sub

　　发布/订阅通信模式是基于主题（Topic）概念的。主题是消息的发布目标（受控对象）。它和队列的不同就在于，可以有多个发送消息和接收消息的客户机，每个主题可以有多个发布者和多个订阅者。

　　JMS API

　　JMS API是在javax.jms包中定义的。要使用JMS API 需要创建一个提供连接对象的连接工厂。连接对象提供与消息服务器的链接。链接被用来创建会话，会话被用来创建消息，消息通过消息的生产者发送到目标（队列或主题），然后消息传递到消息消费者。

 

JMS Parent	PTP Domain	Pub/Sub Domain
ConnectionFactory	QueueConnectionFactory	TopicConnectionFactory
Connection	QueueConnection	TopicConnection
Destination	Queue	Topic
Session	QueueSession	TopicSession
MessageProducer	QueueSender	TopicPublisher
MessageConsumer	QueueReceiver	TopicSubscriber

 

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

　　1、连接工厂（Connection Factory）

　　是客户机用来创建与JMD提供者的链接的对象。它是受控对象，可以通过JNDI查找。JMS API定义了两种类型的连接工厂。 QueueConnectionFactory 和 TopicConnectionFactory

　　2、连接（Connection）

　　连接对象是和JMS提供者通信的媒介。这个通信的具体实现依赖于JMS提供者。除了通用的借口，还有基于队列（QueueConnection）和基于主题（TopicConnection）专用接口。

 

 

 

 

 

3、会话（Session）

　　用于创建消息的生产者、消费者和消息。它们都是单线程，能参加事务。有QueueSession和TopicSession。

　　4、消息（Message）

　　消息是消息服务器在客户端之间发送的一条条信息。有五种接口，不同的类型消息。1. StreamMessage -- Java原始值的数据流  2. MapMessage--一组名/值对  3.TextMessage--一个字符串对象 4. ObjectMessage--一个序列化的 Java对象 5.BytesMessage--一个未解释字节的数据流。

　　消息由以下几部分组成：

　　消息头（header）：JMS消息头包含了许多字段，它们是消息发送后由JMS提供者或消息发送者产生，用来表示消息、设置优先权和失效时间等等，并且为消息确定路由。

　　属性（property）：由消息发送者产生，用来添加删除消息头以外的附加信息。

　　消息体（body）：由消息发送者产生。

　　5、目标（Destination）

　　目标是受控对象。在JMS中表示一个队列或者一个主题。

　　6、消息生产者（MessageProducer）

　　是用于将消息发送到目标的对象，由会话对象创建，有QueueSender、TopicPublisher.

　　7、消息消费者（MessageConsumer）

　　是由会话对象创建，用于从目标获取消息，有QueueReceiver、TopicSubscriber

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

　　1）用JNDI 得到ConnectionFactory对象；

　　2）用ConnectionFactory创建Connection 对象；

　　3）用Connection对象创建一个或多个JMS Session；

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

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

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

 

 

消息生产者程序如下

 

package org.jms.test;
import java.io.*;
mport 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);
        }
    }
}

 

 

 

消息消费者程序如下

 

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;
        }
    }
}