ActiveMQ 笔记

一、简介

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

 JMS即Java消息服务（Java Message Service）应用程序接口是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。Java消息服务是一个与具体平台无关的API，绝大多数MOM提供商都对JMS提供支持。

       在 Java 里有 JMS 的多个实现，ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现。

       JMS 定义了两种方式：Quere（点对点）；Topic（发布/订阅）。

       ConnectionFactory 是连接工厂，负责创建Connection。Connection 负责创建 Session。Destination 是消息的目的地。

       Session 创建 MessageProducer（用来发消息） 和 MessageConsumer（用来接收消息）。

       ActiveMQ的官方网址：http://activemq.apache.org。在此可以下载ActiveMQ的最新版本和阅读相关文档。

二、特性

⒈ 多种语言和协议编写客户端。语言: Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议： OpenWire,Stomp REST,WS Notification,XMPP,AMQP

⒉ 完全支持JMS1.1和J2EE 1.4规范 （持久化，XA消息，事务)

⒊ 对Spring的支持，ActiveMQ可以很容易内嵌到使用Spring的系统里面去，而且也支持Spring2.0的特性

⒋ 通过了常见J2EE服务器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的测试，其中通过JCA 1.5 resource adaptors的配置，可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上

⒌ 支持多种传送协议：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA

⒍ 支持通过JDBC和journal提供高速的消息持久化

⒎ 从设计上保证了高性能的集群，客户端-服务器，点对点

⒏ 支持Ajax

⒐ 支持与Axis的整合

⒑ 可以很容易得调用内嵌JMS provider，进行测试

二、示例

1.消息发送者

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
* 说明： activemq send message
*
* @author xajava
* @version 创建时间：2012-10-24 下午1:22:40
*/
publicclassJmsSender{

privatefinalString USER =ActiveMQConnection.DEFAULT_USER;
privatefinalString PASSWORD =ActiveMQConnection.DEFAULT_PASSWORD;
privatefinalString URL ="tcp://192.168.36.129:61616";//ActiveMQConnection.DEFAULT_BROKER_URL;
privatefinalString SUBJECT ="ActiveMQ.Demo";

privateDestination destination =null;
privateConnection conn =null;
privateSession session =null;
privateMessageProducer producer =null;

// 初始化
privatevoid initialize()throwsJMSException,Exception{
// 连接工厂
ActiveMQConnectionFactory connectionFactory =newActiveMQConnectionFactory(USER, PASSWORD, URL);
conn = connectionFactory.createConnection();
// 事务性会话，自动确认消息
session = conn.createSession(false,Session.AUTO_ACKNOWLEDGE);
// 消息的目的地（Queue/Topic）
destination = session.createQueue(SUBJECT);
// destination = session.createTopic(SUBJECT);
// 消息的提供者（生产者）
producer = session.createProducer(destination);
// 不持久化消息
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
}

publicvoid sendMessage(String msgType)throwsJMSException,Exception{
initialize();
// 连接到JMS提供者（服务器）
conn.start();
// 发送文本消息
if("text".equals(msgType)){
String textMsg ="ActiveMQ Text Message!";
TextMessage msg = session.createTextMessage();
// TextMessage msg = session.createTextMessage(textMsg);
msg.setText(textMsg);
producer.send(msg);
}
// 发送Map消息
if("map".equals(msgType)){
MapMessage msg = session.createMapMessage();
msg.setBoolean("boolean",true);
msg.setShort("short",(short)0);
msg.setLong("long",123456);
msg.setString("MapMessage","ActiveMQ Map Message!");
producer.send(msg);
}
// 发送流消息
if("stream".equals(msgType)){
String streamValue ="ActiveMQ stream Message!";
StreamMessage msg = session.createStreamMessage();
msg.writeString(streamValue);
msg.writeBoolean(false);
msg.writeLong(1234567890);
producer.send(msg);
}
// 发送对象消息
if("object".equals(msgType)){
JmsObjectMessageBean jmsObject =newJmsObjectMessageBean("ActiveMQ Object Message",18,false);
ObjectMessage msg = session.createObjectMessage();
msg.setObject(jmsObject);
producer.send(msg);
}
// 发送字节消息
if("bytes".equals(msgType)){
String byteValue ="字节消息";
BytesMessage msg = session.createBytesMessage();
msg.writeBytes(byteValue.getBytes());
producer.send(msg);
}
}

// 关闭连接
publicvoid close()throwsJMSException{
if(producer !=null)
producer.close();
if(session !=null)
session.close();
if(conn !=null)
conn.close();
}

publicstaticvoid main(String[] args)throwsJMSException,Exception{
JmsSender sender =newJmsSender();

sender.sendMessage("text");
sender.sendMessage("map");
sender.sendMessage("stream");
sender.sendMessage("object");
sender.sendMessage("bytes");
sender.close();

System.out.println("sender test stop");
}
}

2.消息接受者

import java.util.Enumeration;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
* 说明：
*
* @author xajava
* @version 创建时间：2012-10-24 下午2:06:48
*/
publicclassJmsReceiverimplementsMessageListener{

privatefinalString USER =ActiveMQConnection.DEFAULT_USER;
privatefinalString PASSWORD =ActiveMQConnection.DEFAULT_PASSWORD;
privatefinalString URL ="tcp://192.168.36.129:61616";//ActiveMQConnection.DEFAULT_BROKER_URL;
privatefinalString SUBJECT ="ActiveMQ.Demo";

privateDestination dest =null;
privateConnection conn =null;
privateSession session =null;
privateMessageConsumer consumer =null;

privatefinalboolean stop =false;

// 初始化
privatevoid initialize()throwsJMSException,Exception{
// 连接工厂是用户创建连接的对象.
ActiveMQConnectionFactory connectionFactory =newActiveMQConnectionFactory(USER, PASSWORD, URL);
// 连接工厂创建一个jms connection
conn = connectionFactory.createConnection();
// 是生产和消费的一个单线程上下文。会话用于创建消息的生产者，消费者和消息。会话提供了一个事务性的上下文。
session = conn.createSession(false,Session.AUTO_ACKNOWLEDGE);// 不支持事务
// 目的地是客户用来指定他生产消息的目标还有他消费消息的来源的对象.
dest = session.createQueue(SUBJECT);
// dest = session.createTopic(SUBJECT);
// 会话创建消息的生产者将消息发送到目的地
consumer = session.createConsumer(dest);
}

/**
* 消费消息
*
* @throws JMSException
* @throws Exception
*/
publicvoid receiveMessage()throwsJMSException,Exception{
initialize();
conn.start();
consumer.setMessageListener(this);
// 等待接收消息
while(!stop){
Thread.sleep(1000);
// System.out.println("Still alive");
}

}

@SuppressWarnings("rawtypes")
@Override
publicvoid onMessage(Message msg){
try{
if(msg instanceofTextMessage){
TextMessage message =(TextMessage) msg;
System.out.println("------Received TextMessage------");
System.out.println(message.getText());
}elseif(msg instanceofMapMessage){
MapMessage message =(MapMessage) msg;
System.out.println("------Received MapMessage------");
System.out.println(message.getLong("long"));
System.out.println(message.getBoolean("boolean"));
System.out.println(message.getShort("short"));
System.out.println(message.getString("MapMessage"));
System.out.println("------Received MapMessage for while------");
Enumeration enumer = message.getMapNames();
while(enumer.hasMoreElements()){
Object obj = enumer.nextElement();
System.out.println(message.getObject(obj.toString()));
}
}elseif(msg instanceofStreamMessage){
StreamMessage message =(StreamMessage) msg;
System.out.println("------Received StreamMessage------");
System.out.println(message.readString());
System.out.println(message.readBoolean());
System.out.println(message.readLong());
}elseif(msg instanceofObjectMessage){
System.out.println("------Received ObjectMessage------");
ObjectMessage message =(ObjectMessage) msg;
JmsObjectMessageBean jmsObject =(JmsObjectMessageBean) message.getObject();
System.out.println(jmsObject.getUserName()+"__"+ jmsObject.getAge()+"__"+ jmsObject.isFlag());
}elseif(msg instanceofBytesMessage){
System.out.println("------Received BytesMessage------");
BytesMessage message =(BytesMessage) msg;
byte[] byteContent =newbyte[1024];
int length =-1;
StringBuffer content =newStringBuffer();
while((length = message.readBytes(byteContent))!=-1){
content.append(newString(byteContent,0, length));
}
System.out.println(content.toString());
}else{
System.out.println(msg);
}
// stop = true;
}catch(JMSException e){
e.printStackTrace();
}finally{
// try {
// this.close();
// } catch (JMSException e) {
// e.printStackTrace();
// }
}
}

// 关闭连接
publicvoid close()throwsJMSException{
System.out.println("Consumer:->Closing connection");
if(consumer !=null)
consumer.close();
if(session !=null)
session.close();
if(conn !=null)
conn.close();
}

publicstaticvoid main(String[] args)throwsJMSException,Exception{
JmsReceiver receiver =newJmsReceiver();

receiver.receiveMessage();
// receiver.close();

}
}

3.消息对象

import java.io.Serializable;

/**
* 说明： JMS 对象消息示例对象
*
* @author xajava
* @version 创建时间：2012-10-24 下午1:56:07
*/
publicclassJmsObjectMessageBeanimplementsSerializable{

privatestaticfinallong serialVersionUID =2620024932905963095L;

privateString userName;
privateint age =16;
privateboolean flag =true;

publicJmsObjectMessageBean(String userName,int age,boolean flag){
this.setUserName(userName);
this.setAge(age);
this.setFlag(flag);
}

publicString getUserName(){
return userName;
}

publicvoid setUserName(String userName){
this.userName = userName;
}

publicint getAge(){
return age;
}

publicvoid setAge(int age){
this.age = age;
}

publicboolean isFlag(){
return flag;
}

publicvoid setFlag(boolean flag){
this.flag = flag;
}

}

三、整合Spring

问题一：为什么要整合ActiveMQ？

      传统的JDBC代码在处理连接、语句、结果集和异常时是多么冗长和繁杂你一定不会忘记，传统的JMS继承了JDBC的“关荣传统”。发送一个简单的消息，要几行代码呢？请仔细数数吧。

Java代码  

1. public void sendMessage() throws JMSException {  
2.     ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(  
3.     "tcp://localhost:61616");  
4.     Connection connection = null;  
5.     Session session=null;  
6.     try {  
7.           
8.         connection = (Connection) connectionFactory.createConnection();//创建连接  
9.         session = (Session) connection.createSession(false,  
10.                 Session.AUTO_ACKNOWLEDGE);//创建会话  
11.         Destination destination = session.createQueue("myQueue");  
12.         MessageProducer producer = session.createProducer(destination);  
13.         TextMessage message = session.createTextMessage(expectedBody);  
14.         message.setStringProperty("headname", "remoteB");  
15.         producer.send(message);  
16.         connection.close();  
17.     } catch (Exception e) {  
18.         e.printStackTrace();  
19.     }finally{  
20.         try {  
21.             if(session!=null){  
22.                 session.close();  
23.             }  
24.             if(connection!=null){  
25.                 connection=null;  
26.             }  
27.         } catch (Exception e) {  
28.               
29.         }  
30.     }  
31. }  

 

传统接受消息而是类似的代码，其实我们的目的就是发送和接受消息。幸运的是Spring为我们提供了大量的模板。项目一期用的较多的是JdbCTemplate，spring也为我们提供了JMSTemplate模板。

 

问题二：JMSTemplate模板该如何配置呢？

类似于jdbcTemplate，首先要配置一个ConnectionFactory，我们采用ActiveMQ5.2作为消息服务器。之后要开始配置JmsTemplate模板了。最后是配置消息目标了。消息分为队列和主题两大类，因此要配置两个消息目标了。

 

Java代码  

1. <!-- ActiveMQ -->  
2.       
3.     <!-- 配置JMS连接工厂 -->    
4.     <bean id="JmsConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">    
5.         <property name="brokerURL" value="tcp://localhost:61616"/>    
6.     </bean>    
7.     <!-- 配置JMS模版 -->    
8.     <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">    
9.         <property name="connectionFactory" ref="JmsConnectionFactory"/>    
10.     </bean>    
11.     <!-- 发送消息的目的地（队列） -->    
12.     <bean id="QueueDestination" class="org.apache.activemq.command.ActiveMQQueue">    
13.         <!-- 设置消息队列的名字 -->    
14.         <constructor-arg index="0" value="HelloWorldQueue"/>    
15.     </bean>  
16.     <!-- 发送消息的目的地（主题） -->    
17.     <bean id="TopicDestination" class="org.apache.activemq.command.ActiveMQTopic">    
18.         <!-- 设置消息主题的名字 -->    
19.         <constructor-arg index="0" value="FlexTopic"/>    
20.     </bean>     

 

问题三：如何使用JmsTemplate发送消息呢?

spring的beanfactory得到一个jmsTemplate的实例和消息目标的实例，发送消息，够简单的吧。看看代码：

Java代码  

1. JmsTemplate template = (JmsTemplate) SpringContext.getBean("JmsTemplate");  
2.             ActiveMQTopic destination=(ActiveMQTopic)SpringContext.getBean("TopicDestination");  
3.             template.send((javax.jms.Destination) destination, new MessageCreator(){  
4.                 public Message createMessage(Session session) throws JMSException {  
5.                     return session.createTextMessage("hello");  
6.                 }  
7.                   
8.             });  

 问题四：上面的代码能不能在简单些？

很多时候，发送消息的目标都是默认的，因此是不是可以在jmsTemplate中设置一个默认的消息目标呢？答案是肯定的。

Java代码  

1. <!-- 配置JMS模版 -->    
2.    <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">    
3.     <property name="connectionFactory" ref="JmsConnectionFactory"/>  
4.     <property name="defaultDestination" ref="TopicDestination"></property>  
5.    </bean>   

 

发送消息的时候，不指定目标，spring就会调用默认的目标了。

Java代码  

1. JmsTemplate template = (JmsTemplate) SpringContext.getBean("JmsTemplate");  
2.             template.send( new MessageCreator(){  
3.                 public Message createMessage(Session session) throws JMSException {  
4.                     return session.createTextMessage("hello");  
5.                 }  
6.             });  

 

问题四：jmsTemplate怎么接受信息？

 

jmsTemplate接收消息十分的简单，只需要调用template.receive()方法，receive方法是同步的，默认情况下，对receive()方法的调用会造成阻塞，知道消息到达目标----如果必要，永远等下去。为了避免对消息内容等待，可以配置jmsTemplate时，通过设置receiveTimeout属性来指定接收消息超时时间。下面的配置将接收消息的超时时间设置为一分钟（60000毫秒）。

Xml代码  

1. <!-- 配置JMS模版 -->    
2.    <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">    
3.     <property name="connectionFactory" ref="JmsConnectionFactory"/>  
4.     <property name="defaultDestination" ref="TopicDestination"></property>  
5.     <property name="receiveTimeout" value="60000"></property>  
6.    </bean>    

 

 template.receive()会从默认目标接收消息，如果你希望指定一个目标，可以传一个目标。如：template.receive("myQueue").

同步接收消息并不是spring唯一的选择，消息监听器可以实现异步（下篇文章将会介绍消息驱动）。

 

问题五：怎么自动将消息转化为Java对象？

 

转化器在很多组件中都是必不缺少的东西。Spring挺过MessageConverter接口提供了对消息转换的支持。

Java代码  

1. public class MyMessageConverter implements MessageConverter {  
2.     @Override  
3.     public Object fromMessage(Message arg0) throws JMSException,  
4.             MessageConversionException {  
5.         // TODO Auto-generated method stub  
6.         return null;  
7.     }  
8.     @Override  
9.     public Message toMessage(Object arg0, Session arg1) throws JMSException,  
10.             MessageConversionException {  
11.         // TODO Auto-generated method stub  
12.         return null;  
13.     }  
14. }  

 

MessageConverter接口的两个方法简单明了。在发送端toMessage会将java对象转化为消息，在接收端fromMessage会将消息转化为java对象。

下面的代码简单的实现了MessageConverter的两个接口。

Java代码  

1. public class MyMessageConverter implements MessageConverter {  
2.     @Override  
3.     public Object fromMessage(Message message) throws JMSException,  
4.             MessageConversionException {  
5.         if(!(message instanceof MapMessage)){  
6.             throw new MessageConversionException("Messae is not MapMessage");  
7.         }  
8.         MapMessage mapMessage=(MapMessage)message;  
9.         MessageObj messageObj=new MessageObj();  
10.         messageObj.setId(mapMessage.getString("id"));  
11.         messageObj.setInfo(mapMessage.getString("info"));  
12.         return messageObj;  
13.     }  
14.     @Override  
15.     public Message toMessage(Object obj, Session session) throws JMSException,  
16.             MessageConversionException {  
17.         if(!(obj instanceof MessageObj)){  
18.             throw new MessageConversionException("obj is not MessageObj");  
19.         }  
20.         MessageObj messageObj=(MessageObj)obj;  
21.         MapMessage mapMessage=session.createMapMessage();  
22.         mapMessage.setString("id", messageObj.getId());  
23.         mapMessage.setString("info", messageObj.getInfo());  
24.         return mapMessage;  
25.     }  
26. }  

 

此时，发送和接收消息要换成template.convertAndSend(message);template.receiveAndConvert();

可是jmsTemplate如何知道消息转换器呢？需要在配置jmsTemplate的时候，加上messageConverter属性。

Java代码  

1. <!-- 配置JMS模版 -->    
2.    <bean id="JmsTemplate" class="org.springframework.jms.core.JmsTemplate">    
3.     <property name="connectionFactory" ref="JmsConnectionFactory"/>  
4.     <property name="defaultDestination" ref="TopicDestination"></property>  
5.     <property name="receiveTimeout" value="60000"></property>  
6.     <property name="messageConverter" ref="messageObj"></property>  
7.    </bean>    

 messageObj是要转化的java对象的bean的id。

四、原理

1. 存储机制

AMQ Message Store 是什么？

 

默认的activemq消息存储是通过一个所谓的AMQ Message Store来完成。
AMQ Message Store是一个高效的可嵌入支持事务的消息存储解决方案。
在此方案下消息（Message）本身以日志的形式实现持久化，存放在Data Log里。并且还对日志里的消息做了引用索引，方便快速取回Message。

一般情况下消息索引存放于内存（Cache）中，MQ Server定期将索引内容持久化，存放到Reference Store。
Message Data Log文件是有容量限制的，默认是32MB，可自行配置容量。当该Data Log文件里所有消息都被消费完的时候，Data Log文件就会被加上一个标记，通知下一次消息清理时可以被处理掉（处理方式可以是delete或是转移到Achieve目录）。

 

AMQ Message Store方案中 Cache 、Data Log、Reference Store 协作图如下：

 

AMQ Message Store的属性是可以配置的，
你可以在conf/activemq.xml配置文件里添加上如下配置：

 

 

 

 

 

 

 

 

Xml代码  

1. <persistenceAdapter>   
2. <amqPersistenceAdapter directory="activemq-data" maxFileLength="32mb"/>  
3.  </persistenceAdapter>   

 

属性说明

property name	default value	Comments
directory	activemq-data	存储消息文件和日志的目录
useNIO	true	使用 NIO 特性
syncOnWrite	false	同步写文件到磁盘
maxFileLength	32mb	Message Data日志文件的最大 Size
persistentIndex	true	持久化日志索引，如果设为 false ，则在内存中保存
maxCheckpointMessageAddSize	4kb	在自动提交前在事务中能保持的最大消息数
cleanupInterval	30000	每隔多少时间清理不再使用的消息日志（毫秒）
indexBinSize	1024	这个值是用来提升索引的性能的，值越大，索引相对性能越好
indexKeySize	96	index key的size，index key基于message id
indexPageSize	16kb	索引页的size
directoryArchive	archive	消费完的Data Log存放的目录
archiveDataLogs	false	设置为true的话，消费完的Data Log就放到Archive目录，而不是删除。

 

 

AMQ Message Store体系中 目录结构参照下图 ：

 

顶层目录broker name
用broker name命名，默认目录名是localhost，broker name在activemq的配置文件里指定，以下是它的子目录:

 

archive
丢弃的Data Log就放到这里，当archiveDataLogs 属性配置为true时才会存在

 

journal
message data log的所在

 

kr-store
reference store 目录

 

data
引用索引所在目录

 

state
记录store的状态

 

tmp-storage
用来存储一些事物性的消息以减轻内存的负担例如等待正常但是速度很慢的消费端来消费非持久化的Topic.

 

 

其他持久化方式
 activemq同样支持JDBC持久化Message，我们只需要把配置从

 

Xml代码  

1. <persistenceAdapter>   
2. <amqPersistenceAdapter directory="activemq-data" maxFileLength="32mb"/>  
3. </persistenceAdapter>  

 

 
改成AMQ Message Store and JDBC（推荐，同时使用两者可以同时保证效率和可靠性）:

 

Xml代码  

1. <persistenceAdapter>  
2. <journaledJDBC dataDirectory="${activemq.base}/data" dataSource="#oracle-ds"/>  
3. </persistenceAdapter>  

 

 或是 JDBC only（现阶段不推荐，单独的JDBC存储效率不高）：

 

 

Xml代码  

1. <persistenceAdapter>  
2. <jdbcPersistenceAdapter dataSource="#oracle-ds"/>  
3. </persistenceAdapter>  

 

 

Oracle数据源的配置如下：

 

Xml代码  

1. <bean id="oracle-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">  
2.       <property name="driverClassName" value="oracle.jdbc.driver.OracleDriver"/>  
3.       <property name="url" value="jdbc:oracle:thin:@localhost:1521:AMQDB"/>  
4.       <property name="username" value="scott"/>  
5.       <property name="password" value="tiger"/>  
6.       <property name="maxActive" value="200"/>  
7.       <property name="poolPreparedStatements" value="true"/>  
8.     </bean>  

 
其他数据源配置参照Activemq的文档。

2.讯息传送机制以及ACK机制

    AcitveMQ是作为一种消息存储和分发组件，涉及到client与broker端数据交互的方方面面，它不仅要担保消息的存储安全性，还要提供额外的手段来确保消息的分发是可靠的。

 

一. ActiveMQ消息传送机制

    Producer客户端使用来发送消息的， Consumer客户端用来消费消息；它们的协同中心就是ActiveMQ broker,broker也是让producer和consumer调用过程解耦的工具，最终实现了异步RPC/数据交换的功能。随着ActiveMQ的不断发展，支持了越来越多的特性，也解决开发者在各种场景下使用ActiveMQ的需求。比如producer支持异步调用；使用flow control机制让broker协同consumer的消费速率；consumer端可以使用prefetchACK来最大化消息消费的速率；提供"重发策略"等来提高消息的安全性等。在此我们不详细介绍。

 

    一条消息的生命周期如下:

  

     图片中简单的描述了一条消息的生命周期,不过在不同的架构环境中,message的流动行可能更加复杂.将在稍后有关broker的架构中详解..一条消息从producer端发出之后，一旦被broker正确保存，那么它将会被consumer消费，然后ACK，broker端才会删除；不过当消息过期或者存储设备溢出时，也会终结它。

 

 

     这是一张很复杂，而且有些凌乱的图片；这张图片中简单的描述了:1)producer端如何发送消息 2) consumer端如何消费消息 3) broker端如何调度。如果用文字来描述图示中的概念，恐怕一言难尽。图示中，提及到prefetchAck，以及消息同步、异步发送的基本逻辑；这对你了解下文中的ACK机制将有很大的帮助。

 

二. optimizeACK

    "可优化的ACK"，这是ActiveMQ对于consumer在消息消费时，对消息ACK的优化选项，也是consumer端最重要的优化参数之一，你可以通过如下方式开启:

    1) 在brokerUrl中增加如下查询字符串： 

 

String brokerUrl = "tcp://localhost:61616?" + 
                   "jms.optimizeAcknowledge=true" + 
                   "&jms.optimizeAcknowledgeTimeOut=30000" + 
                   "&jms.redeliveryPolicy.maximumRedeliveries=6";
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);

 

    2) 在destinationUri中，增加如下查询字符串：

 

String queueName = "test-queue?customer.prefetchSize";
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination queue = session.createQueue(queueName);

     

    我们需要在brokerUrl指定optimizeACK选项，在destinationUri中指定prefetchSize(预获取)选项，其中brokerUrl参数选项是全局的，即当前factory下所有的connection/session/consumer都会默认使用这些值；而destinationUri中的选项，只会在使用此destination的consumer实例中有效；如果同时指定，brokerUrl中的参数选项值将会被覆盖。optimizeAck表示是否开启“优化ACK”，只有在为true的情况下，prefetchSize(下文中将会简写成prefetch)以及optimizeAcknowledgeTimeout参数才会有意义。此处需要注意"optimizeAcknowledgeTimeout"选项只能在brokerUrl中配置。

    prefetch值建议在destinationUri中指定，因为在brokerUrl中指定比较繁琐；在brokerUrl中，queuePrefetchSize和topicPrefetchSize都需要单独设定："&jms.prefetchPolicy.queuePrefetch=12&jms.prefetchPolicy.topicPrefetch=12"等来逐个指定。

 

    如果prefetchACK为true，那么prefetch必须大于0；当prefetchACK为false时，你可以指定prefetch为0以及任意大小的正数。不过，当prefetch=0是，表示consumer将使用PULL(拉取)的方式从broker端获取消息，broker端将不会主动push消息给client端，直到client端发送PullCommand时；当prefetch>0时，就开启了broker push模式，此后只要当client端消费且ACK了一定的消息之后，会立即push给client端多条消息。

 

    当consumer端使用receive()方法同步获取消息时，prefetch可以为0和任意正值；当prefetch=0时，那么receive()方法将会首先发送一个PULL指令并阻塞，直到broker端返回消息为止，这也意味着消息只能逐个获取(类似于Request<->Response)，这也是Activemq中PULL消息模式；当prefetch > 0时，broker端将会批量push给client 一定数量的消息(<= prefetch),client端会把这些消息(unconsumedMessage)放入到本地的队列中，只要此队列有消息，那么receive方法将会立即返回，当一定量的消息ACK之后，broker端会继续批量push消息给client端。

 

    当consumer端使用MessageListener异步获取消息时，这就需要开发设定的prefetch值必须 >=1,即至少为1；在异步消费消息模式中，设定prefetch=0,是相悖的，也将获得一个Exception。

 

    此外，我们还可以brokerUrl中配置“redelivery”策略，比如当一条消息处理异常时，broker端可以重发的最大次数；和下文中提到REDELIVERED_ACK_TYPE互相协同。当消息需要broker端重发时，consumer会首先在本地的“deliveredMessage队列”(Consumer已经接收但还未确认的消息队列)删除它，然后向broker发送“REDELIVERED_ACK_TYPE”类型的确认指令，broker将会把指令中指定的消息重新添加到pendingQueue(亟待发送给consumer的消息队列)中，直到合适的时机，再次push给client。

 

    到目前为止，或许你知道了optimizeACK和prefeth的大概意义，不过我们可能还会有些疑惑！！optimizeACK和prefetch配合，将会达成一个高效的消息消费模型：批量获取消息，并“延迟”确认(ACK)。prefetch表达了“批量获取”消息的语义，broker端主动的批量push多条消息给client端，总比client多次发送PULL指令然后broker返回一条消息的方式要优秀很多，它不仅减少了client端在获取消息时阻塞的次数和阻塞的时间，还能够大大的减少网络开支。optimizeACK表达了“延迟确认”的语义(ACK时机)，client端在消费消息后暂且不发送ACK，而是把它缓存下来(pendingACK)，等到这些消息的条数达到一定阀值时，只需要通过一个ACK指令把它们全部确认；这比对每条消息都逐个确认，在性能上要提高很多。由此可见，prefetch优化了消息传送的性能，optimizeACK优化了消息确认的性能。

 

    当consumer端消息消费的速率很高(相对于producer生产消息)，而且消息的数量也很大时(比如消息源源不断的生产)，我们使用optimizeACK + prefetch将会极大的提升consumer的性能。不过反过来：

    1) 如果consumer端消费速度很慢(对消息的处理是耗时的)，过大的prefetchSize，并不能有效的提升性能，反而不利于consumer端的负载均衡(只针对queue)；按照良好的设计准则，当consumer消费速度很慢时，我们通常会部署多个consumer客户端，并使用较小的prefetch，同时关闭optimizeACK，可以让消息在多个consumer间“负载均衡”(即均匀的发送给每个consumer)；如果较大的prefetchSize，将会导致broker一次性push给client大量的消息，但是这些消息需要很久才能ACK(消息积压)，而且在client故障时，还会导致这些消息的重发。

 

    2) 如果consumer端消费速度很快，但是producer端生成消息的速率较慢，比如生产者10秒钟生成10条消息，但是consumer一秒就能消费完毕，而且我们还部署了多个consumer！！这种场景下，建议开启optimizeACK，但是需要设置较小的prefetchSize；这样可以保证每个consumer都能有"活干"，否则将会出现一个consumer非常忙碌，但是其他consumer几乎收不到消息。

 

    3) 如果消息很重要，特别是不原因接收到”redelivery“的消息，那么我们需要将optimizeACK=false，prefetchSize=1

 

    既然optimizeACK是”延迟“确认，那么就引入一种潜在的风险：在消息被消费之后还没有来得及确认时，client端发生故障，那么这些消息就有可能会被重新发送给其他consumer，那么这种风险就需要client端能够容忍“重复”消息。

 

    prefetch值默认为1000，当然这个值可能在很多场景下是偏大的；我们暂且不考虑ACK_MODE(参见下文)，通常情况下，我们只需要简单的统计出单个consumer每秒的最大消费消息数即可，比如一个consumer每秒可以处理100个消息，我们期望consumer端每2秒确认一次，那么我们的prefetchSize可以设置为100 * 2 /0.65大概为300。无论如何设定此值，client持有的消息条数最大为：prefetch + “DELIVERED_ACK_TYPE消息条数”(DELIVERED_ACK_TYPE参见下文)

 

     即使当optimizeACK为true，也只会当session的ACK_MODE为AUTO_ACKNOWLEDGE时才会生效，即在其他类型的ACK_MODE时consumer端仍然不会“延迟确认”，即:

consumer.optimizeAck = connection.optimizeACK && session.isAutoAcknowledge()

 

    当consumer.optimizeACK有效时，如果客户端已经消费但尚未确认的消息(deliveredMessage)达到prefetch * 0.65，consumer端将会自动进行ACK；同时如果离上一次ACK的时间间隔，已经超过"optimizeAcknowledgeTimout"毫秒，也会导致自动进行ACK。

 

    此外简单的补充一下，批量确认消息时，只需要在ACK指令中指明“firstMessageId”和“lastMessageId”即可，即消息区间，那么broker端就知道此consumer(根据consumerId识别)需要确认哪些消息。

 
三. ACK模式与类型介绍

    JMS API中约定了Client端可以使用四种ACK_MODE,在javax.jms.Session接口中:

 

• AUTO_ACKNOWLEDGE = 1    自动确认
• CLIENT_ACKNOWLEDGE = 2    客户端手动确认   
• DUPS_OK_ACKNOWLEDGE = 3    自动批量确认
• SESSION_TRANSACTED = 0    事务提交并确认

    此外AcitveMQ补充了一个自定义的ACK_MODE:

• INDIVIDUAL_ACKNOWLEDGE = 4    单条消息确认

    我们在开发JMS应用程序的时候,会经常使用到上述ACK_MODE,其中"INDIVIDUAL_ACKNOWLEDGE "只有ActiveMQ支持,当然开发者也可以使用它. ACK_MODE描述了Consumer与broker确认消息的方式(时机),比如当消息被Consumer接收之后,Consumer将在何时确认消息。对于broker而言，只有接收到ACK指令,才会认为消息被正确的接收或者处理成功了,通过ACK，可以在consumer与Broker之间建立一种简单的“担保”机制. 

   

    Client端指定了ACK_MODE,但是在Client与broker在交换ACK指令的时候,还需要告知ACK_TYPE,ACK_TYPE表示此确认指令的类型，不同的ACK_TYPE将传递着消息的状态，broker可以根据不同的ACK_TYPE对消息进行不同的操作。

 

    比如Consumer消费消息时出现异常,就需要向broker发送ACK指令,ACK_TYPE为"REDELIVERED_ACK_TYPE",那么broker就会重新发送此消息。在JMS API中并没有定义ACT_TYPE,因为它通常是一种内部机制,并不会面向开发者。ActiveMQ中定义了如下几种ACK_TYPE(参看MessageAck类):

 

• DELIVERED_ACK_TYPE = 0    消息"已接收"，但尚未处理结束
• STANDARD_ACK_TYPE = 2    "标准"类型,通常表示为消息"处理成功"，broker端可以删除消息了
• POSION_ACK_TYPE = 1    消息"错误",通常表示"抛弃"此消息，比如消息重发多次后，都无法正确处理时，消息将会被删除或者DLQ(死信队列)
• REDELIVERED_ACK_TYPE = 3    消息需"重发"，比如consumer处理消息时抛出了异常，broker稍后会重新发送此消息
• INDIVIDUAL_ACK_TYPE = 4    表示只确认"单条消息",无论在任何ACK_MODE下    
• UNMATCHED_ACK_TYPE = 5    BROKER间转发消息时,接收端"拒绝"消息

    到目前为止,我们已经清楚了大概的原理: Client端在不同的ACK_MODE时,将意味着在不同的时机发送ACK指令,每个ACK Command中会包含ACK_TYPE,那么broker端就可以根据ACK_TYPE来决定此消息的后续操作. 接下来,我们详细的分析ACK_MODE与ACK_TYPE.

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 

 

    我们需要在创建Session时指定ACK_MODE,由此可见,ACK_MODE将是session共享的,意味着一个session下所有的 consumer都使用同一种ACK_MODE。在创建Session时,开发者不能指定除ACK_MODE列表之外的其他值.如果此session为事务类型,用户指定的ACK_MODE将被忽略,而强制使用"SESSION_TRANSACTED"类型;如果session非事务类型时,也将不能将 ACK_MODE设定为"SESSION_TRANSACTED",毕竟这是相悖的.   

 

 

    Consumer消费消息的风格有2种: 同步/异步..使用consumer.receive()就是同步，使用messageListener就是异步；在同一个consumer中，我们不能使用使用这2种风格，比如在使用listener的情况下，当调用receive()方法将会获得一个Exception。两种风格下，消息确认时机有所不同。

    "同步"伪代码：

 

//receive伪代码---过程
Message message = sessionMessageQueue.dequeue();
if(message != null){
    ack(message);
}
return message

 

    同步调用时，在消息从receive方法返回之前，就已经调用了ACK；因此如果Client端没有处理成功，此消息将丢失(可能重发，与ACK_MODE有关)。

    "异步"伪代码：

 

//基于listener
Session session = connection.getSession(consumerId);
sessionQueueBuffer.enqueue(message);
Runnable runnable = new Ruannale(){
    run(){
        Consumer consumer = session.getConsumer(consumerId);
        Message md = sessionQueueBuffer.dequeue();
        try{
            consumer.messageListener.onMessage(md);
            ack(md);//
        }catch(Exception e){
            redelivery();//sometime，not all the time;
    }
}
//session中将采取线程池的方式，分发异步消息
//因此同一个session中多个consumer可以并行消费
threadPool.execute(runnable);

 

    基于异步调用时，消息的确认是在onMessage方法返回之后，如果onMessage方法异常，会导致消息重发。

 

四. ACK_MODE详解

 

    AUTO_ACKNOWLEDGE : 自动确认,这就意味着消息的确认时机将有consumer择机确认."择机确认"似乎充满了不确定性,这也意味着,开发者必须明确知道"择机确认"的具体时机,否则将有可能导致消息的丢失,或者消息的重复接受.那么在ActiveMQ中,AUTO_ACKNOWLEDGE是如何运作的呢?

    1) 对于consumer而言，optimizeAcknowledge属性只会在AUTO_ACK模式下有效。

 

    2) 其中DUPS_ACKNOWLEGE也是一种潜在的AUTO_ACK,只是确认消息的条数和时间上有所不同。

 

    3) 在“同步”(receive)方法返回message之前,会检测optimizeACK选项是否开启，如果没有开启，此单条消息将立即确认，所以在这种情况下，message返回之后，如果开发者在处理message过程中出现异常，会导致此消息也不会redelivery,即"潜在的消息丢失"；如果开启了optimizeACK，则会在unAck数量达到prefetch * 0.65时确认，当然我们可以指定prefetchSize = 1来实现逐条消息确认。

 

    4) 在"异步"(messageListener)方式中,将会首先调用listener.onMessage(message),此后再ACK,如果onMessage方法异常,将导致client端补充发送一个ACK_TYPE为REDELIVERED_ACK_TYPE确认指令；如果onMessage方法正常,消息将会正常确认(STANDARD_ACK_TYPE)。此外需要注意，消息的重发次数是有限制的，每条消息中都会包含“redeliveryCounter”计数器，用来表示此消息已经被重发的次数，如果重发次数达到阀值，将会导致发送一个ACK_TYPE为POSION_ACK_TYPE确认指令,这就导致broker端认为此消息无法消费,此消息将会被删除或者迁移到"dead letter"通道中。

    

    因此当我们使用messageListener方式消费消息时，通常建议在onMessage方法中使用try-catch,这样可以在处理消息出错时记录一些信息，而不是让consumer不断去重发消息；如果你没有使用try-catch,就有可能会因为异常而导致消息重复接收的问题,需要注意你的onMessage方法中逻辑是否能够兼容对重复消息的判断。

 

 
 
 
 

    CLIENT_ACKNOWLEDGE : 客户端手动确认，这就意味着AcitveMQ将不会“自作主张”的为你ACK任何消息，开发者需要自己择机确认。在此模式下，开发者需要需要关注几个方法：1) message.acknowledge()，2) ActiveMQMessageConsumer.acknowledege()，3) ActiveMQSession.acknowledge()；其1)和3)是等效的，将当前session中所有consumer中尚未ACK的消息都一起确认，2)只会对当前consumer中那些尚未确认的消息进行确认。开发者可以在合适的时机必须调用一次上述方法。

 

    我们通常会在基于Group(消息分组)情况下会使用CLIENT_ACKNOWLEDGE，我们将在一个group的消息序列接受完毕之后确认消息(组)；不过当你认为消息很重要，只有当消息被正确处理之后才能确认时，也很可以使用此ACK_MODE。

 

    如果开发者忘记调用acknowledge方法，将会导致当consumer重启后，会接受到重复消息，因为对于broker而言，那些尚未真正ACK的消息被视为“未消费”。

    开发者可以在当前消息处理成功之后，立即调用message.acknowledge()方法来"逐个"确认消息，这样可以尽可能的减少因网络故障而导致消息重发的个数；当然也可以处理多条消息之后，间歇性的调用acknowledge方法来一次确认多条消息，减少ack的次数来提升consumer的效率，不过这仍然是一个利弊权衡的问题。

 

    除了message.acknowledge()方法之外，ActiveMQMessageConumser.acknowledge()和ActiveMQSession.acknowledge()也可以确认消息，只不过前者只会确认当前consumer中的消息。其中sesson.acknowledge()和message.acknowledge()是等效的。

 

    无论是“同步”/“异步”，ActiveMQ都不会发送STANDARD_ACK_TYPE，直到message.acknowledge()调用。如果在client端未确认的消息个数达到prefetchSize * 0.5时，会补充发送一个ACK_TYPE为DELIVERED_ACK_TYPE的确认指令，这会触发broker端可以继续push消息到client端。(参看PrefetchSubscription.acknwoledge方法)

 

    在broker端，针对每个Consumer，都会保存一个因为"DELIVERED_ACK_TYPE"而“拖延”的消息个数，这个参数为prefetchExtension，事实上这个值不会大于prefetchSize * 0.5,因为Consumer端会严格控制DELIVERED_ACK_TYPE指令发送的时机(参见ActiveMQMessageConsumer.ackLater方法)，broker端通过“prefetchExtension”与prefetchSize互相配合，来决定即将push给client端的消息个数，count = prefetchExtension + prefetchSize - dispatched.size()，其中dispatched表示已经发送给client端但是还没有“STANDARD_ACK_TYPE”的消息总量；由此可见，在CLIENT_ACK模式下，足够快速的调用acknowledge()方法是决定consumer端消费消息的速率；如果client端因为某种原因导致acknowledge方法未被执行，将导致大量消息不能被确认，broker端将不会push消息，事实上client端将处于“假死”状态，而无法继续消费消息。我们要求client端在消费1.5*prefetchSize个消息之前，必须acknowledge()一次；通常我们总是每消费一个消息调用一次，这是一种良好的设计。

 

    此外需要额外的补充一下：所有ACK指令都是依次发送给broker端，在CLIET_ACK模式下，消息在交付给listener之前，都会首先创建一个DELIVERED_ACK_TYPE的ACK指令，直到client端未确认的消息达到"prefetchSize * 0.5"时才会发送此ACK指令，如果在此之前，开发者调用了acknowledge()方法，会导致消息直接被确认(STANDARD_ACK_TYPE)。broker端通常会认为“DELIVERED_ACK_TYPE”确认指令是一种“slow consumer”信号，如果consumer不能及时的对消息进行acknowledge而导致broker端阻塞，那么此consumer将会被标记为“slow”，此后queue中的消息将会转发给其他Consumer。

 

    DUPS_OK_ACKNOWLEDGE : "消息可重复"确认，意思是此模式下，可能会出现重复消息，并不是一条消息需要发送多次ACK才行。它是一种潜在的"AUTO_ACK"确认机制，为批量确认而生，而且具有“延迟”确认的特点。对于开发者而言，这种模式下的代码结构和AUTO_ACKNOWLEDGE一样，不需要像CLIENT_ACKNOWLEDGE那样调用acknowledge()方法来确认消息。

 

    1) 在ActiveMQ中，如果在Destination是Queue通道，我们真的可以认为DUPS_OK_ACK就是“AUTO_ACK + optimizeACK + (prefetch > 0)”这种情况，在确认时机上几乎完全一致；此外在此模式下，如果prefetchSize =1 或者没有开启optimizeACK，也会导致消息逐条确认，从而失去批量确认的特性。

 

    2) 如果Destination为Topic，DUPS_OK_ACKNOWLEDGE才会产生JMS规范中诠释的意义，即无论optimizeACK是否开启，都会在消费的消息个数>=prefetch * 0.5时，批量确认(STANDARD_ACK_TYPE),在此过程中，不会发送DELIVERED_ACK_TYPE的确认指令,这是1)和AUTO_ACK的最大的区别。

 

    这也意味着，当consumer故障重启后，那些尚未ACK的消息会重新发送过来。

 

    SESSION_TRANSACTED : 当session使用事务时，就是使用此模式。在事务开启之后，和session.commit()之前，所有消费的消息，要么全部正常确认，要么全部redelivery。这种严谨性，通常在基于GROUP(消息分组)或者其他场景下特别适合。在SESSION_TRANSACTED模式下，optimizeACK并不能发挥任何效果,因为在此模式下，optimizeACK会被强制设定为false，不过prefetch仍然可以决定DELIVERED_ACK_TYPE的发送时机。

 

    因为Session非线程安全，那么当前session下所有的consumer都会共享同一个transactionContext；同时建议，一个事务类型的Session中只有一个Consumer，已避免rollback()或者commit()方法被多个consumer调用而造成的消息混乱。

    

    当consumer接受到消息之后，首先检测TransactionContext是否已经开启，如果没有，就会开启并生成新的transactionId，并把信息发送给broker；此后将检测事务中已经消费的消息个数是否 >= prefetch * 0.5,如果大于则补充发送一个“DELIVERED_ACK_TYPE”的确认指令；这时就开始调用onMessage()方法，如果是同步(receive),那么即返回message。上述过程，和其他确认模式没有任何特殊的地方。

   

    当开发者决定事务可以提交时，必须调用session.commit()方法，commit方法将会导致当前session的事务中所有消息立即被确认；事务的确认过程中，首先把本地的deliveredMessage队列中尚未确认的消息全部确认(STANDARD_ACK_TYPE)；此后向broker发送transaction提交指令并等待broker反馈，如果broker端事务操作成功，那么将会把本地deliveredMessage队列清空，新的事务开始；如果broker端事务操作失败(此时broker已经rollback)，那么对于session而言，将执行inner-rollback，这个rollback所做的事情，就是将当前事务中的消息清空并要求broker重发(REDELIVERED_ACK_TYPE),同时commit方法将抛出异常。

 

    当session.commit方法异常时，对于开发者而言通常是调用session.rollback()回滚事务(事实上开发者不调用也没有问题)，当然你可以在事务开始之后的任何时机调用rollback(),rollback意味着当前事务的结束，事务中所有的消息都将被重发。需要注意，无论是inner-rollback还是调用session.rollback()而导致消息重发，都会导致message.redeliveryCounter计数器增加，最终都会受限于brokerUrl中配置的"jms.redeliveryPolicy.maximumRedeliveries",如果rollback的次数过多，而达到重发次数的上限时，消息将会被DLQ(dead letter)。

 

    INDIVIDUAL_ACKNOWLEDGE : 单条消息确认，这种确认模式，我们很少使用，它的确认时机和CLIENT_ACKNOWLEDGE几乎一样，当消息消费成功之后，需要调用message.acknowledege来确认此消息(单条)，而CLIENT_ACKNOWLEDGE模式先message.acknowledge()方法将导致整个session中所有消息被确认(批量确认)。

 

    结语：到目前为止，我们已经已经简单的了解了ActiveMQ中消息传送机制，还有JMS中ACK策略，重点分析了optimizeACK的策略，希望开发者能够在使用activeMQ中避免一些不必要的错误。本文如有疏漏和错误之处，请各位不吝赐教，特此感谢。

五、注意

1 使用jms需要注意的问题

一下所述的问题，不仅是对ActiveMQ，对于其他的JMS也一样有效。

1.1 不要频繁的建立和关闭连接

JMS使用长连接方式，一个程序，只要和JMS服务器保持一个连接就可以了，不要频繁的建立和关闭连接。频繁的建立和关闭连接，对程序的性能影响还是很大的。这一点和jdbc还是不太一样的。

1.2 Connection的start()和stop()方法代价很高

JMS 的Connection的start()和stop()方法代价很高，不能经常调用。我们试用的时候，写了个jms的connection pool，每次将connection取出pool时调用start()方法，归还时调用stop()方法，然而后来用jprofiler发现，一般的 cpu时间都耗在了这两个方法上。

1.3 start()后才能收消息

Connection的start()方法调用后，才能收到jms消息。如果不调用这个方法，能发出消息，但是一直收不到消息。不知道其它的jms服务器也是这样。

1.4 显式关闭Session

如 果忘记了最后关闭Connection或Session对象，都会导致内存泄漏。这个在我测试的时候也发现了。本来以为关闭了Connection，由这 个Connection生成的Session也会被自动关闭，结果并非如此，Session并没有关闭，导致内存泄漏。所以一定要显式的关闭 Connection和Session。

1.5 对Session做对象池

对Session做对象池，而不是 Connection。Session也是昂贵的对象，每次使用都新建和关闭，代价也非常高。而且后来我们发现，原来Connection是线程安全的， 而Session不是，所以后来改成了对Session做对象池，而只保留一个Connection。

2 集群

ActiveMQ有强大而灵活的集群功能，但是使用起来还是会有很多陷阱。

2.1 broker cluster和 master-slave

ActiveMQ 可以做broker的集群，也可以做master-slave方式的集群。前者能在多个broker之前fail-over和load-balance， 但是在某个节点出故障时，可能导致消息丢失；而后者能实时备份消息，和fail-over，但是不能load-balance。broker cluser的方式，在一个broker上发送的消息可以在其它的broker上收到。当一个broker失效时，客户端可以自动的转到别的broker 上运行，多个broker可以同时提供服务，但是消息只存储在一个broker上，如果那个broker失效了，那么客户端直到它重新启动后才能收到该 broker上的消息，假如很不幸，那个broker的存储介质坏了，那么消息就丢失掉了。
Master-slave方式中，只有master提供服务，slave只是实时的备份master的数据，所以消息不会丢失。当master失效 时，slave会自动升为master，客户端会自动转到slave上工作，所以能fail-over。由于只有master提供服务，所以不能将负载分 到多个broker上。
其实单个broker的性能已经是相当的惊人了，在我们公司的机器上能达到每秒收发4000个消息，没个消息4K字节这样的速度，足够公司目前的需要了，而公司并不希望丢失任何数据，所以我们选择使用master-slave模式。

2.2 多种master-slave模式

master-slave也有多种实现方式。它们的不同只是在共享数据和锁机制上。

2.2.1 Pure master-slave

Pure master-slave，显示的在配置文件中指定一个broker做为另一个broker的slave。运行时，slave同过网络自动从master 出复制数据，同时在和master失去连接时自动升级为master。当master失效，slave成为master后，如果要让原先的master重 新投入运行，需要停掉运行中的slave(现在升级为master了)，手动复制slave中的数据到master中。再重新启动master和 slave。这种方式最简单，效率也不错，但是只能有两台做集群，只能fail-over一次，而且需要停机回复master-slave结构。

2.2.2 JDBC master-slave

这 种方式不需要特殊的配置，只要让所有的节点都把数据存储到同一个数据库中。先拿到数据库表的锁的节点成为master，一旦它失效了，其它的节点获得锁， 就可以成为master。因为数据通过数据库共享，放在一个地方，不需要停机恢复master-slave。这种方式，需要额外的数据库服务器，如果数据 库失效了，那么就全失效了，而且速度不是很快。我们在用mysql测试时，并没有成功，master失效后，其他的节点始终没有升级成slave，可能是 数据库配置的问题。

2.2.3 Share file master-slave

这种方式类似于前者，也不需要特别的配置，只是通过共享文件系统来共享数据，靠文件锁实现只有一台成为master。共享文件系统的方式有很多，我们测试了nfs v4 (v3有bug，不行)， 最终在稳定性，效率等方面不是很满意，可能是通过网络太慢了。

测 试过众多master-slave模式后发现，pure方式管理起来麻烦，jdbc方式成本高，效率低，share file方式需要高性能的共享文件，都有缺点。鉴于单台activeMQ很可靠，而我们的基础平台组愿意用硬件备份，最终还是决定不用master- slave了，也不用broker cluster，就用单台，通过硬件冗余保证数据不会丢失，并找另外一台刀片机做冷备，在主服务器失效时顶替。

 

http://sulong.me/archives/207

 

比 如说：订单创建后，发送一条消息到一个队列，另一个专门发邮件的程序接受队列里消息，给用户发送邮件。事实上，所有需要给用户发邮件的地方都只要给这个队 列里发送消息就可以了。 这样做的好处在于，用户不用等到邮件发送结束，就可以先看到订单创建结果。整个系统中只有一处对立发邮件功能，其它应用都只要向这个应用发消息就行了。

 

异 步调用的本质是可以让调用者在被调用者完成之前返回，这样调用者不用等待被调用者完成调用。所以异步的第一个用途就是用来提高响应速度。比如我前面的例子 里，用户确认创建订单后，他只要知道订单被创建好了就行了，订单创建后的发邮件等一系列的事情是否发生，有没有成功执行，他不关心。用户在感觉上就是响应 速度提升了。

至于你说的可扩展性，伸缩性，等等，都得建立在他们可以异步的基础上。比如，用户登录功能，用户在提交过登录信息后，就得马上见到是否登录成功，你不能告诉他你正在登录中，半个小时候来看结果。这个时候就算异步可以提升XX性，也是不能采用的。

再 次，采用异步后，并不意味着就能在扩展性，伸缩性或运行效率上的提升。扩展性主要是靠良好的程序设计来实现。在良好的程序设计的基础上，你才容易在不改动 或较少改动原有代码的基础上编写出新功能。如果程序设计得不好，即便用了异步，也不容易添加新的功能。异步也只能提升系统中那些可以异步的应用的伸缩性， 并不一定能提升整个系统的伸缩性。比方说，如果发送邮件服务器的负载太高，我们可以添加新的服务器来监听消息发邮件，发送邮件的应用伸缩性是提高了；如果 创建订单应用由于设计上的原因不能通过简单的加入服务器就可以支持更多的负载，那么创建订单应用的伸缩性并没有因为异步发送邮件而有多大的改善。对于整个 系统来说，伸缩性并没有太大的改善。最后，运行效率有很多具体的指标，我觉得异步确实提升了响应速度，但是对其它的指标可能也没有多少改善。

六、资源

ActiveMQ_百度百科ActiveMQ 视频课程 - JAVA - 项目学院 - 浪曦网—中国首家IT原创视频在线学习，教学，下载, 分享集一体教育平台ActiveMQ发消息和收消息 - 叶子的日志 - 网易博客ActiveMQ消息发送和接收 - xajava - BlogJavaSpring整和ActiveMQ - flex爱好者 - ITeye技术网站activemq的消息存储机制 - 幸福的懦夫 - ITeye技术网站ActiveMQ讯息传送机制以及ACK机制详解_软件架构设计大全_优良自学吧activemq 使用经验 - 兽魔（英雄、创新、实用、稳定） - ITeye技术网站Geronimo 中 JMS、MDB 和 ActiveMQ 的使用技巧ActiveMQ_全部文档 - OPEN开源文档