ActiveMQ总结.doc

1. 背景
当前，CORBA、DCOM、RMI等RPC中间件技术已广泛应用于各个领域。但是面对规模和复杂度都越来越高的分布式系统，这些技术也显示出其局限性：（1）同步通信：客户发出调用后，必须等待服务对象完成处理并返回结果后才能继续执行；（2）客户和服务对象的生命周期紧密耦合：客户进程和服务对象进程 都必须正常运行；如果由于服务对象崩溃或者网络故障导致客户的请求不可达，客户会接收到异常；（3）点对点通信：客户的一次调用只发送给某个单独的目标对象。
面向消息的中间件（Message Oriented Middleware，MOM）较好的解决了以上问题。发送者将消息发送给消息服务器，消息服务器将消息存放在若干队列中，在合适的时候再将消息转发给接收者。这种模式下，发送和接收是异步的，发送者无需等待；二者的生命周期未必相同：发送消息的时候接收者不一定运行，接收消息的时候发送者也不一定运行； 一对多通信：对于一个消息可以有多个接收者。
已有的MOM系统包括IBM的MQSeries、Microsoft的MSMQ和BEA的MessageQ等。由于没有一个通用的标准，这些系统很难实现互操作和无缝连接。Java Message Service（JMS）是SUN提出的旨在统一各种MOM系统接口的规范，它包含点对点（Point to Point，PTP）和发布/订阅（Publish/Subscribe，pub/sub）两种消息模型，提供可靠消息传输、事务和消息过滤等机制。
2．JMS概述
2.1 JMS规范
JAVA 消息服务(JMS)定义了Java 中访问消息中间件的接口。JMS 只是接口，并没有给予实现，实现JMS 接口的消息中间件称为JMS Provider，例如ActiveMQ。
2.2 术语
JMS Provider：实现JMS 接口的消息中间件；
PTP：Point to Point，即点对点的消息模型；
Pub/Sub：Publish/Subscribe，即发布/订阅的消息模型；
Queue：队列目标；
Topic：主题目标；
ConnectionFactory：连接工厂，JMS 用它创建连接；
Connection：JMS 客户端到JMS Provider 的连接；
Destination：消息的目的地；
Session：会话，一个发送或接收消息的线程；
MessageProducer：由Session 对象创建的用来发送消息的对象；
MessageConsumer：由Session 对象创建的用来接收消息的对象;
Acknowledge：签收；
Transaction：事务。
2.3 JMS编程模型
在 JMS 编程模型中，JMS 客户端（组件或应用程序）通过 JMS 消息服务交换消息。消息生产者将消息发送至消息服务，消息消费者则从消息服务接收这些消息。这些消息传送操作是使用一组实现 JMS 应用编程接口 (API) 的对象（由 JMS Provide提供）来执行的。
在 JMS 编程模型中，JMS 客户端使用 ConnectionFactory 对象创建一个连接，向消息服务发送消息以及从消息服务接收消息均是通过此连接来进行。Connection 是客户端与消息服务的活动连接。创建连接时，将分配通信资源以及验证客户端。这是一个相当重要的对象，大多数客户端均使用一个连接来进行所有的消息传送。
连接用于创建会话。Session 是一个用于生成和使用消息的单线程上下文。它用于创建发送的生产者和接收消息的消费者，并为所发送的消息定义发送顺序。会话通过大量确认选项或通过事务来支持可靠传送。
客户端使用 MessageProducer 向指定的物理目标（在 API 中表示为目标身份对象）发送消息。生产者可指定一个默认传送模式（持久性消息与非持久性消息）、优先级和有效期值，以控制生产者向物理目标发送的所有消息。
同样，客户端使用 MessageConsumer 对象从指定的物理目标（在 API 中表示为目标对象）接收消息。消费者可使用消息选择器，借助它，消息服务可以只向消费者发送与选择标准匹配的那些消息。
消费者可以支持同步或异步消息接收。异步使用可通过向消费者注册 MessageListener 来实现。当会话线程调用 MessageListener 对象的 onMessage 方法时，客户端将使用消息。
2.4 JMS编程域
JMS 支持两种截然不同的消息传送模型：PTP（即点对点模型）和Pub/Sub（即发布/订阅模型），分别称作：PTP Domain 和Pub/Sub Domain。
PTP（使用Queue即队列目标） 消息从一个生产者传送至一个消费者。在此传送模型中，目标是一个队列。消息首先被传送至队列目标，然后根据队列传送策略，从该队列将消息传送至向此队列进行注册的某一个消费者，一次只传送一条消息。可以向队列目标发送消息的生产者的数量没有限制，但每条消息只能发送至、并由一个消费者成功使用。如果没有已经向队列目标注册的消费者，队列将保留它收到的消息，并在某个消费者向该队列进行注册时将消息传送给该消费者。
Pub/Sub（使用Topic即主题目标） 消息从一个生产者传送至任意数量的消费者。在此传送模型中，目标是一个主题。消息首先被传送至主题目标，然后传送至所有已订阅此主题的活动消费者。可以向主题目标发送消息的生产者的数量没有限制，并且每个消息可以发送至任意数量的订阅消费者。主题目标也支持持久订阅的概念。持久订阅表示消费者已向主题目标进行注册，但在消息传送时此消费者可以处于非活动状态。当此消费者再次处于活动状态时，它将接收此信息。如果没有已经向主题目标注册的消费者，主题不保留其接收到的消息，除非有非活动消费者注册了持久订阅。
这两种消息传送模型使用表示不同编程域的 API 对象（其语义稍有不同）进行处理，如下所示：
基本类型（统一域）
PTP域
Pub/Sub域

ConnectionFactory
QueueConnectionFactory
TopicConnectionFactory

Connection
QueueConnection
TopicConnection

Session
QueueSession
TopicPublisher

Destination（Queue或 Topic）
Queue
Topic

MessageProducer
QueueSender

MessageConsumer
QueueReceiver,QueueBrowser
TopicSubscriber

使用图表第一列中列出的统一域对象编写点对点和发布/订阅消息传送。这是首选方法（JMS 1.1规范）。然而，为了符合早期的 JMS 1.02b 规范，可以使用PTP域对象编写点对点消息传送，使用Pub/Sub域对象编制发布/订阅消息传送。
2.5 JMS消息结构
JMS 消息由以下几部分组成：消息头，属性和消息体。
2.5.1 消息头(Header)
消息头包含消息的识别信息和路由信息，消息头包含一些标准的属性如：JMSDestination，JMSMessageID等。
消息头
由谁设置

JMSDestination
send方法

JMSDeliveryMode
send方法

JMSExpiration
send方法

JMSPriority
send方法

JMSMessageID
send方法

JMSTimestamp
send方法

JMSCorrelationID
客户

JMSReplyTo
客户

JMSType
客户

JMSRedelivered
JMS Provider

2.5.2 属性(Properties)
除了消息头中定义好的标准属性外，JMS 提供一种机制增加新属性到消息头中，这种新属性包含以下几种：
1. 应用需要用到的属性;
2. 消息头中原有的一些可选属性;
3. JMS Provider 需要用到的属性。
标准的JMS 消息头包含以下属性：
消息头
描述

JMSDestination
消息发送的目的地

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

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

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

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

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

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

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

JMSType
消息类型的识别符。

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

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

TextMessage
java.lang.String对象，如xml文件内容

MapMessage
名/值对的集合，名是String对象，值类型可以是Java任何基本类型

BytesMessage
字节流

StreamMessage
Java中的输入输出流

ObjectMessage
Java中的可序列化对象

Message
没有消息体，只有消息头和属性

2.6 PTP模型
PTP(Point-to-Point)模型是基于队列的，生产者发消息到队列，消费者从队列接收消息，队列的存在使得消息的异步传输成为可能。和邮件系统中的邮箱一样，队列可以包含各种消息，JMS Provider 提供工具管理队列的创建、删除。JMS PTP 模型定义了客户端如何向队列发送消息，从队列接收消息，浏览队列中的消息。
下面描述JMS PTP 模型中的主要概念和对象：
名称
描述

ConnectionFactory
客户端用ConnectionFactory 创建Connection 对象。

Connection
一个到JMS Provider 的连接，客户端可以用Connection 创建Session 来发送和接收消息。

Session
客户端用Session 创建MessageProducer 和MessageConsumer对象。如果在Session 关闭时，有一些消息已经被收到，但还没有被签收(acknowledged)，那么，当消费者下次连接到相同的队列时，这些消息还会被再次接收。

Destination（Queue或 TemporaryQueue）
客户端用Session 创建Destination 对象。此处的目标为队列，队列由队列名识别。临时队列只能由创建它的Connection 所创建的消费者消费，但是任何生产者都可向临时队列发送消息。

MessageProducer
客户端用MessageProducer 发送消息到队列。

MessageConsumer
客户端用MessageConsumer 接收队列中的消息，如果用户在receive方法中设定了消息选择条件，那么不符合条件的消息会留在队列中，不会被接收到。

可靠性(Reliability)
队列可以长久地保存消息直到消费者收到消息。消费者不需要因为担心消息会丢失而时刻和队列保持激活的连接状态，充分体现了异步传输模式的优势。

2.7 PUB/SUB模型
JMS Pub/Sub 模型定义了如何向一个内容节点发布和订阅消息，这些节点被称作主题(topic)。
主题可以被认为是消息的传输中介，发布者(publisher)发布消息到主题，订阅者(subscribe) 从主题订阅消息。主题使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。
下面描述JMS Pub/Sub 模型中的主要概念和对象：
名称
描述

订阅(subscription)
消息订阅分为非持久订阅(non-durable subscription)和持久订阅(durable subscrip-tion)，非持久订阅只有当客户端处于激活状态，也就是和JMS Provider 保持连接状态才能收到发送到某个主题的消息，而当客户端处于离线状态，这个时间段发到主题的消息将会丢失，永远不会收到。持久订阅时，客户端向JMS 注册一个识别自己身份的ID，当这个客户端处于离线时，JMS Provider 会为这个ID 保存所有发送到主题的消息，当客户再次连接到JMS Provider时，会根据自己的ID 得到所有当自己处于离线时发送到主题的消息。

ConnectionFactory
客户端用ConnectionFactory 创建Connection 对象。

Connection
一个到JMS Provider 的连接，客户端可以用Connection 创建Session 来发送和接收消息。

Session
客户端用Session 创建MessageProducer 和MessageConsumer对象。它还提供持久订阅主题，或使用unsubscribe 方法取消消息的持久订阅。

Destination（Topic和TemporaryTopic）
客户端用Session 创建Destination 对象。此处的目标为主题，主题由主题名识别。临时主题只能由创建它的Connection所创建的消费者消费。临时主题不能提供持久订阅功能。JMS 没有给出主题的组织和层次结构的定义，由JMS Provider 自己定义。

MessageProducer
客户端用MessageProducer发布消息到主题。

MessageConsumer
客户端用MessageConsumer 接收发布到主题上的消息。可以在receive 中设置消息过滤功能，这样，不符合要求的消息不会被接收。

恢复和重新派送(Recovery and Redelivery)
非持久订阅状态下，不能恢复或重新派送一个未签收的消息。只有持久订阅才能恢复或重新派送一个未签收的消息。

可靠性(Reliability)
当所有的消息必须被接收，则用持久订阅模式。当丢失消息能够被容忍，则用非持久订阅模式。

2.8 JMS支持并发
JMS对象
是否支持并发

Destination
是

ConnectionFactory
是

Connection
是

Session
否

MessageProducer
否

MessageConsumer
否

3. ActiveMQ安装
3.1 版本
jdk版本：jdk1.5.0_11
ActiveMQ版本：ActiveMQ 4.2测试版
C++客户端版本：ActiveMQ CPP 1.1 Release
3.2 ActiveMQ二进制安装
gunzip apache-activemq-4.2-20070328.130210-35.tar.gz
tar xvf apache-activemq-4.2-20070328.130210-35.tar
设置ActiveMQ环境变量：ACTIVEMQ_HOME=安装目录
设置CLASSPATH环境变量, CLASSPATH=$CLASSPATH:$ACTIVEMQ_HOME/ apache-activemq-4.2-SNAPSHOT.jar
3.3 ActiveMQ移植
只需将$ACTIVEMQ_HOME打包移植到新机器即可。
3.4 C++客户端编译
安装perl-5.8.8.tar.gz
tar xzvf perl-5.8.8.tar.gz
cd perl-5.8.8
rm -f config.sh Policy.sh
sh Configure -de -Dprefix=/usr
make
make test
make install
reboot
系统重新启动，登录系统后可以执行 perl -v 查看Perl版本信息
检查/usr/bin/perl或/usr/local/bin/perl是否指向新版本的perl
安装m4-1.4.8.tar.gz
tar xzvf m4-1.4.8.tar.gz
cd m4-1.4.8
./configure
make
make install
安装autoconf-2.59.tar.gz
tar xzvf autoconf-2.59.tar.gz
cd autoconf-2.59
./configure
make
make install
安装automake-1.9.6.tar.gz
tar xzvf automake-1.9.6.tar.gz
cd automake-1.9.6
./configure
make
make install
安装libtool-1.5.22.tar.gz
tar xzvf libtool-1.5.22.tar.gz
cd libtool-1.5.22
./configure
make
make install
安装cppunit-1.10.2.tar.gz
tar xzvf cppunit-1.10.2.tar.gz
cd cppunit-1.10.2
./configure
make
make install
安装e2fsprogs-1.38.tar.gz
注：此项仅Solaris8需要安装，Solaris9已自带此uuid头文件和库文件
tar xzvf e2fsprogs-1.38.tar.gz
cd e2fsprogs-1.38
./configure
make
make install
安装gcc-3.4.6
注：无需自行编译，直接从下载对应solaris版本和cpu的gcc-3.4.6包文件即可
gunzip gcc-3.4.6-sol8-sparc-local.gz
pkgadd –d gcc-3.4.6-sol8-sparc-local
安装activemq-cpp-1.1.tar.gz
tar xzvf activemq-cpp-1.1.tar.gz
cd activemq-cpp-1.1
./autogen.sh
./configure
make
make install
复制activemq-cpp-1.1/include/activemq-cpp-1.1下文件至/usr/include
复制activemq-cpp-1.1/lib/libactivemq-cpp.a至/usr/lib
3.5 C++客户端移植
1．在新机器安装gcc-3.4.6；
2．复制activemq-cpp-1.1/include/activemq-cpp-1.1下文件至新机器/usr/include，复制activemq-cpp-1.1/lib/ libactivemq-cpp.a至新机器/usr/lib；
如果新机器为Solaris8还需如下操作：复制在Solaris8下编译e2fsprogs-1.38.tar.gz产生的uuid头文件和库文件至新机器相应的目录（即/usr/include/uuid/uuid.h和/usr/lib/ libuuid.a，此处复制需注意文件层次）
3.6 启动
cd $ACTIVEMQ_HOME/bin
./activemq
或者
./activemq > activemq.log 2>&1 &
3.6 停止
ps –ef|grep activemq
kill 进程号
4. ActiveMQ编程
ActiveMQ特色：
Supports a variety of Cross Language Clients and Protocols from Java, C, C++, C#, Ruby, Perl, Python, PHP
OpenWire for high performance clients in Java, C, C++, C#
Stomp support so that clients can be written easily in C, Ruby, Perl, Python, PHP to talk to ActiveMQ as well as any other popular Message Broker
Supports many advanced features such as Message Groups, Virtual Destinations, Wildcards and Composite Destinations
Fully supports JMS 1.1 and J2EE 1.4 with support for transient, persistent, transactional and XA messaging
Spring Support so that ActiveMQ can be easily embedded into Spring applications and configured using Spring's XML configuration mechanism
Tested inside popular J2EE servers such as Geronimo, JBoss 4, GlassFish and WebLogic
Includes JCA 1.5 resource adaptors for inbound & outbound messaging so that ActiveMQ should auto-deploy in any J2EE 1.4 compliant server
Supports pluggable transport protocols such as in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports
Supports very fast persistence using JDBC along with a high performance journal
Designed for high performance clustering, client-server, peer based communication
REST API to provide technology agnostic and language neutral web based API to messaging
Ajax to support web streaming support to web browsers using pure DHTML, allowing web browsers to be part of the messaging fabric
Axis Support so that ActiveMQ can be easily dropped into Apache Axis runtimes to provide reliable messaging
Can be used as an in memory JMS provider, ideal for unit testing JMS

本文档仅描述最基本的使用方法，实际使用过程中请以官方文档为准。
1．ActiveMQ官方网站：http://activemq.apache.org
2．JMS官方网站：
4.1 开发JSM的步骤
广义上说，一个JMS 应用是几个JMS 客户端交换消息，开发JMS 客户端应用由以下几步构成：
用JNDI 得到ConnectionFactory 对象；
用ConnectionFactory 创建Connection 对象；
用Connection 对象创建一个或多个JMS Session；
用JNDI 得到目标队列或主题对象，即Destination 对象；
用Session 和Destination 创建MessageProducer 和MessageConsumer；
通知Connection 开始传送消息。
4.2 编程模版
4.2.1 ConnectionFactory
要初始化JMS，则需要使用连接工厂。客户端通过创建ConnectionFactory建立到ActveMQ的连接，一个连接工厂封装了一组连接配置参数，这组参数在配置ActiveMQ时已经定义，例如brokerURL参数，此参数传入的是ActiveMQ服务地址和端口，支持openwire协议的默认连接为tcp://localhost:61616，支持stomp协议的默认连接为tcp://localhost:61613。
注：由于C++客户端暂时仅支持stomp协议，所以需要使用tcp://localhost:61613。
ConnectionFactory支持并发。
Java客户端：
ActiveMQConnectionFactory构造方法：
ActiveMQConnectionFactory();
ActiveMQConnectionFactory(String brokerURL);
ActiveMQConnectionFactory(String userName, String password, String brokerURL) ;
ActiveMQConnectionFactory(String userName, String password, URI brokerURL) ;
ActiveMQConnectionFactory(URI brokerURL);
其中brokerURL为ActiveMQ服务地址和端口。
例如：
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.135:61616");
或者
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory. setBrokerURL("tcp://192.168.0.135:61616");
C++客户端：
ActiveMQConnectionFactory构造函数：
ActiveMQConnectionFactory(void)；
ActiveMQConnectionFactory( const std::string& url,
const std::string& username = "",
const std::string& password = "",
const std::string& clientId = "" );
例如：
ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.0.135:61613");
或者
ActiveMQConnectionFactory* connectionFactory = new ActiveMQConnectionFactory();
connectionFactory-> setBrokerURL("tcp://192.168.0.135:61613");
4.2.2 Connection
在成功创建正确的ConnectionFactory后，下一步将是创建一个连接，它是JMS定义的一个接口。ConnectionFactory负责返回可以与底层消息传递系统进行通信的Connection实现。通常客户端只使用单一连接。根据JMS文档，Connection的目的是“利用JMS提供者封装开放的连接”，以及表示“客户端与提供者服务例程之间的开放TCP/IP套接字”。该文档还指出Connection应该是进行客户端身份验证的地方，除了其他一些事项外，客户端还可以指定惟一标志符。
当一个Connection被创建时，它的传输默认是关闭的，必须使用start方法开启。
一个Connection可以建立一个或多个的Session。
当一个程序执行完成后，必须关闭之前创建的Connection，否则ActiveMQ不能释放资源，关闭一个Connection同样也关闭了Session，MessageProducer和MessageConsumer。
Connection支持并发。
4.2.2.1 创建Connection
Java客户端：
ActiveMQConnectionFactory方法：
Connection createConnection()；
Connection createConnection(String userName, String password);
例如：
Connection connection = connectionFactory.createConnection();
C++客户端：
函数原型：
cms::Connection* ActiveMQConnectionFactory::createConnection(void)
throw ( cms::CMSException )；
cms::Connection* ActiveMQConnectionFactory::createConnection(
const std::string& username,
const std::string& password,
const std::string& clientId )
throw ( cms::CMSException );
例如：
Connection* connection = connectionFactory->createConnection();
4.2.2.2 开启Connection
Java客户端：
ActiveMQConnection方法：
void start();
例如：
Connection.start();
C++客户端：
函数原型：
void ActiveMQConnection::start(void) throw ( cms::CMSException );
例如：
connection->start();
4.2.2.3 关闭Connection
Java客户端：
ActiveMQConnection方法：
void close();
例如：
Connection.close();
C++客户端：
函数原型：
void ActiveMQConnection::close(void) throw ( cms::CMSException );
例如：
connection->close();
4.2.3 Session
一旦从ConnectionFactory中获得一个Connection，就必须从Connection中创建一个或者多个Session。Session是一个发送或接收消息的线程，可以使用Session创建MessageProducer，MessageConsumer和Message。
Session可以被事务化，也可以不被事务化，通常，可以通过向Connection上的适当创建方法传递一个布尔参数对此进行设置。
Java客户端：
ActiveMQConnection方法：
Session createSession(boolean transacted, int acknowledgeMode);
其中transacted为使用事务标识，acknowledgeMode为签收模式。
例如：
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
C++客户端：
函数原型：
cms::Session* ActiveMQConnection::createSession(void);
cms::Session* ActiveMQConnection::createSession(
cms::Session::AcknowledgeMode ackMode );
例如：
Session* session = connection->createSession( Session::AUTO_ACKNOWLEDGE );
4.2.4 Destination
Destination是一个客户端用来指定生产消息目标和消费消息来源的对象。
在PTP模式中，Destination被称作Queue即队列；在Pub/Sub模式，Destination被称作Topic即主题。在程序中可以使用多个Queue和Topic。
Java客户端：
ActiveMQSession方法：
Queue createQueue(String queueName);
Temporar