ActiveMQ - JMS,Transport,Persistence-阿里云开发者社区

关键字: JMS

1 JMS
在介绍ActiveMQ之前，首先简要介绍一下JMS规范。
1．1 JMS的基本构件
1．1．1 连接工厂
连接工厂是客户用来创建连接的对象，例如ActiveMQ提供的ActiveMQConnectionFactory。

1．1．2 连接
JMS Connection封装了客户与JMS提供者之间的一个虚拟的连接。

1．1．3 会话
JMS Session是生产和消费消息的一个单线程上下文。会话用于创建消息生产者（producer）、消息消费者（consumer）和消息（message）等。会话提供了一个事务性的上下文，在这个上下文中，一组发送和接收被组合到了一个原子操作中。

1．1．4 目的地
目的地是客户用来指定它生产的消息的目标和它消费的消息的来源的对象。JMS1.0.2规范中定义了两种消息传递域：点对点（PTP）消息传递域和发布/订阅消息传递域。
点对点消息传递域的特点如下：

每个消息只能有一个消费者。
消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态，它都可以提取消息。

发布/订阅消息传递域的特点如下：

每个消息可以有多个消费者。
生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它订阅之后发布的消息。JMS规范允许客户创建持久订阅，这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。

在点对点消息传递域中，目的地被成为队列（queue）；在发布/订阅消息传递域中，目的地被成为主题（topic）。

1．1．5 消息生产者
消息生产者是由会话创建的一个对象，用于把消息发送到一个目的地。

1．1．6 消息消费者
消息消费者是由会话创建的一个对象，它用于接收发送到目的地的消息。消息的消费可以采用以下两种方法之一：

同步消费。通过调用消费者的receive方法从目的地中显式提取消息。receive方法可以一直阻塞到消息到达。
异步消费。客户可以为消费者注册一个消息监听器，以定义在消息到达时所采取的动作。

1．1．7 消息
JMS消息由以下三部分组成：

消息头。每个消息头字段都有相应的getter和setter方法。
消息属性。如果需要除消息头字段以外的值，那么可以使用消息属性。
消息体。JMS定义的消息类型有TextMessage、MapMessage、BytesMessage、StreamMessage和ObjectMessage。

1．2 JMS的可靠性机制
1．2．1 确认
JMS消息只有在被确认之后，才认为已经被成功地消费了。消息的成功消费通常包含三个阶段：客户接收消息、客户处理消息和消息被确认。
在事务性会话中，当一个事务被提交的时候，确认自动发生。在非事务性会话中，消息何时被确认取决于创建会话时的应答模式（acknowledgement mode）。该参数有以下三个可选值：

Session.AUTO_ACKNOWLEDGE。当客户成功的从receive方法返回的时候，或者从MessageListener.onMessage方法成功返回的时候，会话自动确认客户收到的消息。
Session.CLIENT_ACKNOWLEDGE。客户通过消息的acknowledge方法确认消息。需要注意的是，在这种模式中，确认是在会话层上进行：确认一个被消费的消息将自动确认所有已被会话消费的消息。例如，如果一个消息消费者消费了10个消息，然后确认第5个消息，那么所有10个消息都被确认。
Session.DUPS_ACKNOWLEDGE。该选择只是会话迟钝第确认消息的提交。如果JMS provider失败，那么可能会导致一些重复的消息。如果是重复的消息，那么JMS provider必须把消息头的JMSRedelivered字段设置为true。

1．2．2 持久性
JMS 支持以下两种消息提交模式：

PERSISTENT。指示JMS provider持久保存消息，以保证消息不会因为JMS provider的失败而丢失。
NON_PERSISTENT。不要求JMS provider持久保存消息。

1．2．3 优先级
可以使用消息优先级来指示JMS provider首先提交紧急的消息。优先级分10个级别，从0（最低）到9（最高）。如果不指定优先级，默认级别是4。需要注意的是，JMS provider并不一定保证按照优先级的顺序提交消息。

1．2．4 消息过期
可以设置消息在一定时间后过期，默认是永不过期。

1．2．5 临时目的地
可以通过会话上的createTemporaryQueue方法和createTemporaryTopic方法来创建临时目的地。它们的存在时间只限于创建它们的连接所保持的时间。只有创建该临时目的地的连接上的消息消费者才能够从临时目的地中提取消息。

1．2．6 持久订阅
    首先消息生产者必须使用PERSISTENT提交消息。客户可以通过会话上的createDurableSubscriber方法来创建一个持久订阅，该方法的第一个参数必须是一个topic。第二个参数是订阅的名称。
    JMS provider会存储发布到持久订阅对应的topic上的消息。如果最初创建持久订阅的客户或者任何其它客户使用相同的连接工厂和连接的客户ID、相同的主题和相同的订阅名再次调用会话上的createDurableSubscriber方法，那么该持久订阅就会被激活。JMS provider会象客户发送客户处于非激活状态时所发布的消息。
    持久订阅在某个时刻只能有一个激活的订阅者。持久订阅在创建之后会一直保留，直到应用程序调用会话上的unsubscribe方法。

1．2．7 本地事务
    在一个JMS客户端，可以使用本地事务来组合消息的发送和接收。JMS Session接口提供了commit和rollback方法。事务提交意味着生产的所有消息被发送，消费的所有消息被确认；事务回滚意味着生产的所有消息被销毁，消费的所有消息被恢复并重新提交，除非它们已经过期。
    事务性的会话总是牵涉到事务处理中，commit或rollback方法一旦被调用，一个事务就结束了，而另一个事务被开始。关闭事务性会话将回滚其中的事务。
需要注意的是，如果使用请求/回复机制，即发送一个消息，同时希望在同一个事务中等待接收该消息的回复，那么程序将被挂起，因为知道事务提交，发送操作才会真正执行。
    需要注意的还有一个，消息的生产和消费不能包含在同一个事务中。

1．3 JMS 规范的变迁
JMS的最新版本的是1.1。它和同1.0.2版本之间最大的差别是，JMS1.1通过统一的消息传递域简化了消息传递。这不仅简化了JMS API，也有利于开发人员灵活选择消息传递域，同时也有助于程序的重用和维护。
以下是不同消息传递域的相应接口：

JMS 公共	点对点域	发布/订阅域
ConnectionFactory	QueueConnectionFactory	TopicConnectionFactory
Connection	QueueConnection	TopicConnection
Destination	Queue	Topic
Session	QueueSession	TopicSession
MessageProducer	QueueSender	TopicPublisher
MessageConsumer	QueueReceiver	TopicSubscriber

2 ActiveMQ
2．1 Broker
2．1．1 Running Broker
ActiveMQ5.0 的二进制发布包中bin目录中包含一个名为activemq的脚本，直接运行这个脚本就可以启动一个broker。
此外也可以通过Broker Configuration URI或Broker XBean URI对broker进行配置，以下是一些命令行参数的例子：

Example	Description
activemq	Runs a broker using the default 'xbean:activemq.xml' as the broker configuration file.
activemq xbean:myconfig.xml	Runs a broker using the file myconfig.xml as the broker configuration file that is located in the classpath.
activemq xbean:file:./conf/broker1.xml	Runs a broker using the file broker1.xml as the broker configuration file that is located in the relative file path ./conf/broker1.xml
activemq xbean:file:C:/ActiveMQ/conf/broker2.xml	Runs a broker using the file broker2.xml as the broker configuration file that is located in the absolute file path C:/ActiveMQ/conf/broker2.xml
activemq broker:(tcp://localhost:61616, tcp://localhost:5000)?useJmx=true	Runs a broker with two transport connectors and JMX enabled.
activemq broker:(tcp://localhost:61616, network:tcp://localhost:5000)?persistent=false	Runs a broker with 1 transport connector and 1 network connector with persistence disabled.

2．1．2 Embedded Broker
可以通过在应用程序中以编码的方式启动broker，例如：

    Java代码 
     
   
BrokerService broker = new BrokerService();   
 broker.addConnector("tcp://localhost:61616");    
broker.start();  

        如果需要启动多个broker，那么需要为broker设置一个名字。例如：
   

    Java代码 
     
   
BrokerService broker = new BrokerService();   
 broker.setName("fred");    
 broker.addConnector("tcp://localhost:61616");    
broker.start();  

        如果希望在同一个JVM内访问这个broker，那么可以使用VM Transport，URI是：vm://brokerName。关于更多的broker属性，可以参考Apache的官方文档。
   

此外，也可以通过BrokerFactory来创建broker，例如：

    Java代码 
     
BrokerService broker = BrokerFactory.createBroker(new URI(someURI));

        someURI的可选值如下：
   

URI scheme	Example	Description
xbean:	xbean:activemq.xml	Searches the classpath for an XML document with the given URI (activemq.xml in this case) which will then be used as the Xml Configuration
file:	file:foo/bar/activemq.xml	Loads the given file (in this example foo/bar/activemq.xml) as the Xml Configuration
broker:	broker:tcp://localhost:61616	Uses the Broker Configuration URI to configure the broker

当使用XBean的配置方式的时候，需要指定一个xml配置文件，例如：

    Java代码 
     
BrokerService broker = BrokerFactory.createBroker(new URI("xbean:com/test/activemq.xml"));

        使用Spring的配置方式如下：
   

    Xml代码 
     
   
<bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean">  
   <property name="config" value="classpath:org/apache/activemq/xbean/activemq.xml" />   
   <property name="start" value="true" />   
 </bean>   

2．1．3 Monitoring Broker
2．1．3．1 JMX
在使用JMX监控broker之前，首先要启用broker的JMX监控功能，例如在配置文件中设置useJmx="true"，如下：

    Xml代码 
     
   
<broker useJmx="true" brokerName="broker1>  
   <managementContext>   
      <managementContext createConnector="true"/>   
   </managementContext>   
  ...   
 </broker>   

        接下来运行JDK自带的jconsole。在运行了jconsole后，它会弹出对话框来选择需要连接到的agent。如果是在启动broker的主机上运行jconsole，那么ActiveMQ broker会出现在jconsole的Local 标签中。如果要连接到远程的broker，那么可以在Advanced标签中指定JMX URL，以下是一个连接到本机的JMX URL：
   

service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi

在jconsole的MBeans标签中，可以查看详细信息，也可以执行相应的operation。需要注意的是，在jconsole连接到broker的时候，并不需要输入用户名和密码，如果这存在潜在的安全问题，那么就需要为JMX Connector配置密码保护（需要使用1.5以上版本的JDK）。

首先要禁止ActiveMQ创建自己的connector，例如：

    Xml代码 
     
   
<broker xmlns="http://activemq.org/config/1.0" brokerName="localhost"useJmx="true">  
   <managementContext>   
      <managementContext createConnector="false"/>   
   </managementContext>   
 </broker>   

        然后在ActiveMQ的conf目录下创建一个访问控制文件和密码文件，如下：
   

conf/jmx.access：
# The "monitorRole" role has readonly access.
# The "controlRole" role has readwrite access.
monitorRole readonly
controlRole readwrite

conf/jmx.password：
# The "monitorRole" role has password "abc123".
# The "controlRole" role has password "abcd1234".
monitorRole abc123
controlRole abcd1234

然后修改ActiveMQ的bin目录下activemq的启动脚本，查找包含"SUNJMX="的一行如下：

REM set SUNJMX=-Dcom.sun.management.jmxremote.port=1616 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
把它替换成
set SUNJMX=-Dcom.sun.management.jmxremote.port=1616 -Dcom.sun.management.jmxremote.authenticate=true -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.password.file=%ACTIVEMQ_BASE%/conf/jmx.password -Dcom.sun.management.jmxremote.access.file=%ACTIVEMQ_BASE%/conf/jmx.access
最后重启ActiveMQ和jconsole，这时候需要强制login。如果在启动activemq的过程中出现以下错误，那么需要为这个文件增加访问控制。Windows平台上的具体解决方法请参考如下网址：http://java.sun.com/j2se/1.5.0/docs/guide/management/security-windows.html
Error: Password file read access must be restricted: D:\apache-activemq-5.0.0\bin\../conf/jmx.password

2．1．3．2 Web Console
Web Console被集成到了ActiveMQ的二进制发布包中，因此缺省访问http://localhost:8161/admin即可访问Web Console。
在配置文件中，可以通过修改nioConnector的port属性来修改Web console的缺省端口：

    Xml代码 
     
   
<jetty xmlns="http://mortbay.com/schemas/jetty/1.0">  
   <connectors>   
     <nioConnector port="8161" />   
   </connectors>   
  ...   
 </jetty>   

        出于安全性或者可靠性的考虑，Web Console 可以被部署到不同于ActiveMQ的进程中。例如把activemq-web-console.war部署到一个单独的web容器中（Tomcat，Jetty等）。在ActiveMQ5.0的二进制发布包中不包含activemq-web-console.war，因此需要下载ActiveMQ的源码，然后进入到${activemq.base}/src/activemq-web-console目录中执行mvn instanll。如果一切正常，那么缺省会在${activemq.base}/src/activemq-web-console/target目录中生成activemq-web-console-5.0.0.war。然后将activemq-web-console-5.0.0.war拷贝到Tomcat的webapps目录中，并重命名成activemq-web-console.war。
   

需要注意的是，要将activemq-all-5.0.0.jar拷贝到WEB-INF\lib目录中（可能还需要拷贝jms.jar）。还要为Tomcat设置以下五个系统属性（修改catalina.bat文件）：

set JAVA_OPTS=%JAVA_OPTS% -Dwebconsole.type="properties"
set JAVA_OPTS=%JAVA_OPTS% -Dwebconsole.jms.url="tcp://localhost:61616"
set JAVA_OPTS=%JAVA_OPTS% -Dwebconsole.jmx.url="service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi"
set JAVA_OPTS=%JAVA_OPTS% -Dwebconsole.jmx.role=""
set JAVA_OPTS=%JAVA_OPTS% -Dwebconsole.jmx.password=""

如果JMX没有配置密码保护，那么webconsole.jmx.role和webconsole.jmx.password设置成""即可。如果broker被配置成了Master/Slave模式，那么可以配置成使用failover transport，例如：

-Dwebconsole.jms.url=failover:(tcp://serverA:61616,tcp://serverB:61616)

顺便说一下，由于webconsole.type 属性是properties，因此实际上起作用的Web Console的配置文件是WEB-INF/ webconsole-properties.xml。最后启动被监控的ActiveMQ，访问http://localhost:8080/activemq-web-console/，查看显示是否正常。

2．1．3．3 Advisory Message
ActiveMQ 支持Advisory Messages，它允许你通过标准的JMS 消息来监控系统。目前的Advisory Messages支持：

consumers, producers and connections starting and stopping
temporary destinations being created and destroyed
messages expiring on topics and queues
brokers sending messages to destinations with no consumers.
connections starting and stopping

Advisory Messages可以被想象成某种的管理通道，通过它你可以得到关于JMS provider、producers、consumers和destinations的信息。Advisory topics都使用ActiveMQ.Advisory.这个前缀，以下是目前支持的topics：

Client based advisories

Advisory Topics	Description
ActiveMQ.Advisory.Connection	Connection start & stop messages
ActiveMQ.Advisory.Producer.Queue	Producer start & stop messages on a Queue
ActiveMQ.Advisory.Producer.Topic	Producer start & stop messages on a Topic
ActiveMQ.Advisory.Consumer.Queue	Consumer start & stop messages on a Queue
ActiveMQ.Advisory.Consumer.Topic	Consumer start & stop messages on a Topic

在消费者启动/停止的Advisory Messages的消息头中有个consumerCount属性，他用来指明目前desination上活跃的consumer的数量。

Destination and Message based advisories

Advisory Topics	Description
ActiveMQ.Advisory.Queue	Queue create & destroy
ActiveMQ.Advisory.Topic	Topic create & destroy
ActiveMQ.Advisory.TempQueue	Temporary Queue create & destroy
ActiveMQ.Advisory.TempTopic	Temporary Topic create & destroy
ActiveMQ.Advisory.Expired.Queue	Expired messages on a Queue
ActiveMQ.Advisory.Expired.Topic	Expired messages on a Topic
ActiveMQ.Advisory.NoConsumer.Queue	No consumer is available to process messages being sent on a Queue
ActiveMQ.Advisory.NoConsumer.Topic	No consumer is available to process messages being sent on a Topic

以上的这些destnations都可以用来作为前缀，在其后面追加其它的重要信息，例如topic、queue、clientID、producderID和consumerID等。这令你可以利用Wildcards 和 Selectors 来过滤Advisory Messages（关于Wildcard和Selector会在稍后介绍）。

例如，如果你希望订阅FOO.BAR这个queue上Consumer的start/stop的消息，那么可以订阅ActiveMQ.Advisory.Consumer.Queue.FOO.BAR；如果希望订阅所有queue上的start/stop消息，那么可以订阅ActiveMQ.Advisory.Consumer.Queue.>；如果希望订阅所有queue或者topic上的start/stop消息，那么可以订阅ActiveMQ.Advisory.Consumer. >。

org.apache.activemq.advisory.AdvisorySupport类上有如下的helper methods，用来在程序中得到advisory destination objects。

    Java代码 
     
   
AdvisorySupport.getConsumerAdvisoryTopic()   
AdvisorySupport.getProducerAdvisoryTopic()   
AdvisorySupport.getDestinationAdvisoryTopic()   
AdvisorySupport.getExpiredTopicMessageAdvisoryTopic()   
AdvisorySupport.getExpiredQueueMessageAdvisoryTopic()   
AdvisorySupport.getNoTopicConsumersAdvisoryTopic()   
AdvisorySupport.getNoQueueConsumersAdvisoryTopic()  

       以下是段使用Advisory Messages的程序代码：
   

    Java代码 
     
   
Destination advisoryDestination = AdvisorySupport.getProducerAdvisoryTopic(destination)   
MessageConsumer consumer = session.createConsumer(advisoryDestination);   
 consumer.setMessageListener(this);    
...   
 public void onMessage(Message msg){    
     if (msg instanceof ActiveMQMessage){    
         try {    
             ActiveMQMessage aMsg =  (ActiveMQMessage)msg;   
             ProducerInfo prod = (ProducerInfo) aMsg.getDataStructure();   
         } catch (JMSException e) {    
             log.error("Failed to process message: " + msg);    
        }   
    }   
}  

2．1．3．4 Command Agent
在介绍Command Agent前首先简要介绍一下XMPP(Jabber)协议，XMPP是一种基于XML的即时通信协议，它由Jabber软件基金会开发。在配置文件中通过增加transportConnector来支持XMPP协议：

    Xml代码 
     
   
<broker xmlns="http://activemq.org/config/1.0">  
   <transportConnectors>   
     ...   
      <transportConnector name="xmpp"     uri="xmpp://localhost:61222"/>   
   </transportConnectors>   
 </broker>   

ActiveMQ提供了ActiveMQ messages和XMPP之间的双向桥接：

如果客户加入了一个聊天室，那么这个聊天室的名字会被映射到一个JMS topic。
尝试在聊天室内发送消息会导致一个JMS消息被发送到这个topic。
呆在一个聊天室中意味着这将保持一个对相应JMS topic的订阅。因此发送到这个topic的JMS消息也会被发送到聊天室。

推荐XMPP客户端Spark(http://www.igniterealtime.org/)。

从4.2版本起，ActiveMQ支持Command Agent。在配置文件中，通过设置commandAgent来启用Command Agent：

    Xml代码 
     
   
<beans>  
   <broker useJmx="true" xmlns="http://activemq.org/config/1.0">   
    ...   
   </broker>   
   <commandAgent xmlns="http://activemq.org/config/1.0"/>   
 </beans>   

        启用了Command Agent的broker上会有一个来自Command Agent的连接，它同时订阅topic： ActiveMQ.Agent。在你启动XMPP客户端，加入到ActiveMQ.Agent聊天室后，就可以同broker进行交谈了。通过在XMPP客户端中键入help，可以得到帮助信息。
   

需要注意的是，ActiveMQ5.0版本有个小bug，如果broker没有采用缺省的用户名和密码，那么Command Agent便无法正常启动。Apache官方文档说，此bug已经被修正，预定在5.2.0版本上体现。修改方式如下：

    Xml代码 
     
<commandAgent xmlns="http://activemq.org/config/1.0" brokerUser="user" brokerPassword="passward"/>

2．1．3．5 Visualization plugin
ActiveMQ支持以broker插件的形式生成DOT文件(可以用agrviewer来查看)，以图表的方式描述connections、sessions、producers、consumers、destinations等信息。配置方式如下：

    Xml代码 
     
   
<broker xmlns="http://activemq.org/config/1.0" brokerName="localhost" useJmx="true">  
      ...   
       <plugins>   
           <connectionDotFilePlugin  file="connection.dot"/>   
           <destinationDotFilePlugin file="destination.dot"/>   
       </plugins>   
 </broker>   

       需要注意的是，笔者认为ActiveMQ5.0版本的Visualization Plugin尚不稳定，存在诸多问题。例如：如果使用connectionDotFilePlugin，那么brokerName必须是localhost；如果使用destinationDotFilePlugin可能会导致ArrayStoreException。
   

 
关键字: Transport
2．2 Transport
     ActiveMQ目前支持的transport有：VM Transport、TCP Transport、SSL Transport、Peer Transport、UDP Transport、Multicast Transport、HTTP and HTTPS Transport、Failover Transport、Fanout Transport、Discovery Transport、ZeroConf Transport等。以下简单介绍其中的几种，更多请参考Apache官方文档。
 
2．2．1 VM Transport
     VM transport允许在VM内部通信，从而避免了网络传输的开销。这时候采用的连接不是socket连接，而是直接地方法调用。 第一个创建VM 连接的客户会启动一个embed VM broker，接下来所有使用相同的broker name的VM连接都会使用这个broker。当这个broker上所有的连接都关闭的时候，这个broker也会自动关闭。
     以下是配置语法：
   vm://brokerName?transportOptions
   例如：vm://broker1?marshal=false&broker.persistent=false
   Transport Options的可选值如下：

        
         Option Name 
         Default Value 
         Description 
        
 
         Marshal 
         false 
         If true, forces each command sent over the transport to be marshlled and unmarshlled using a WireFormat 
        
 
         wireFormat 
         default 
         The name of the WireFormat to use 
        
 
         wireFormat.* 
           
         All the properties with this prefix are used to configure the wireFormat 
        
 
         create 
         true 
         If the broker should be created on demand if it does not allready exist. Only supported in ActiveMQ 4.1 
        
 
         broker.* 
           
         All the properties with this prefix are used to configure the broker. See Configuring Wire Formats for more information 
        

       
   以下是高级配置语法：
   vm:(broker:(tcp://localhost)?brokerOptions)?transportOptions
   vm:broker:(tcp://localhost)?brokerOptions
    例如：vm:(broker:(tcp://localhost:6000)?persistent=false)?marshal=false
    Transport Options的可选值如下：

        
         Option Name 
         Default Value 
         Description 
        
 
         marshal 
         false 
         If true, forces each command sent over the transport to be marshlled and unmarshlled using a WireFormat 
        
 
         wireFormat 
         default 
         The name of the WireFormat to use 
        
 
         wireFormat.* 
           
         All the properties with this prefix are used to configure the wireFormat 
        

       
   使用配置文件的配置语法：   
     vm://localhost?brokerConfig=xbean:activemq.xml
     例如：vm:// localhost?brokerConfig=xbean:com/test/activemq.xml
 
   使用Spring的配置：

         Xml代码 
          
        
<bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean">  
   <property name="config" value="classpath:org/apache/activemq/xbean/activemq.xml" />   
   <property name="start" value="true" />   
 </bean>   
  
 <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="broker">   
   <property name="brokerURL" value="vm://localhost"/>   
 </bean>   

            如果persistent是true，那么ActiveMQ会在当前目录下创建一个缺省值是activemq-data的目录用于持久化保存数据。需要注意的是，如果程序中启动了多个不同名字的VM broker，那么可能会有如下警告：Failed to start jmx connector: Cannot bind to URL [rmi://localhost:1099/jmxrmi]: javax.naming.NameAlreadyBoundException…可以通过在transportOptions中追加broker.useJmx=false来禁用JMX来避免这个警告。
        
 
2．2．2 TCP Transport
     TCP transport 允许客户端通过TCP socket连接到远程的broker。以下是配置语法：
     tcp://hostname:port?transportOptions
     Transport Options的可选值如下：

        
         Option Name 
         Default Value 
         Description 
        
 
         minmumWireFormatVersion 
         0 
         The minimum version wireformat that is allowed 
        
 
         trace 
         false 
         Causes all commands that are sent over the transport to be logged 
        
 
         useLocalHost 
         true 
         When true, it causes the local machines name to resolve to "localhost". 
        
 
         socketBufferSize 
         64 * 1024 
         Sets the socket buffer size in bytes 
        
 
         soTimeout 
         0 
         sets the socket timeout in milliseconds 
        
 
         connectionTimeout 
         30000 
         A non-zero value specifies the connection timeout in milliseconds. A zero value means wait forever for the connection to be established. Negative values are ignored. 
        
 
         wireFormat 
         default 
         The name of the WireFormat to use 
        
 
         wireFormat.* 
           
         All the properties with this prefix are used to configure the wireFormat. See Configuring Wire Formats for more information 
        

         例如：tcp://localhost:61616?trace=false
 
2．2．3 Failover Transport
     Failover Transport是一种重新连接的机制，它工作于其它transport的上层，用于建立可靠的传输。它的配置语法允许制定任意多个复合的URI。Failover transport会自动选择其中的一个URI来尝试建立连接。如果没有成功，那么会选择一个其它的URI来建立一个新的连接。以下是配置语法：
     failover:(uri1,...,uriN)?transportOptions
     failover:uri1,...,uriN
     Transport Options的可选值如下：

        
         Option Name 
          Default Value  
         Description 
        
 
         initialReconnectDelay 
         10 
         How long to wait before the first reconnect attempt (in ms) 
        
 
         maxReconnectDelay 
         30000 
         The maximum amount of time we ever wait between reconnect attempts (in ms) 
        
 
         useExponentialBackOff 
         true 
         Should an exponential backoff be used between reconnect attempts 
        
 
         backOffMultiplier 
         2 
         The exponent used in the exponential backoff attempts 
        
 
         maxReconnectAttempts 
         0 
         If not 0, then this is the maximum number of reconnect attempts before an error is sent back to the client 
        
 
         randomize 
         true 
         use a random algorithm to choose the URI to use for reconnect from the list provided 
        
 
         backup 
         false 
         initialize and hold a second transport connection - to enable fast failover 
        

         例如：failover:(tcp://localhost:61616,tcp://remotehost:61616)?initialReconnectDelay=100
 
2．2．4 Discovery transport
     Discovery transport是可靠的tranport。它使用Discovery transport来定位用来连接的URI列表。以下是配置语法：
     discovery:(discoveryAgentURI)?transportOptions
     discovery:discoveryAgentURI
     Transport Options的可选值如下：

        
         Option Name 
         Default Value 
         Description 
        
 
         initialReconnectDelay 
         10 
         How long to wait before the first reconnect attempt 
        
 
         maxReconnectDelay 
         30000 
         The maximum amount of time we ever wait between reconnect attempts 
        
 
         useExponentialBackOff 
         true 
         Should an exponential backoff be used btween reconnect attempts 
        
 
         backOffMultiplier 
         2 
         The exponent used in the exponential backoff attempts 
        
 
         maxReconnectAttempts 
         0 
         If not 0, then this is the maximum number of reconnect attempts before an error is sent back to the client 
        

         例如：discovery:(multicast://default)?initialReconnectDelay=100   
     为了使用Discovery来发现broker，需要为broker启用discovery agent。 以下是XML配置文件中的一个例子：

         Xml代码 
          
        
<broker name="foo">  
    <transportConnectors>   
       <transportConnector uri="tcp://localhost:0" discoveryUri="multicast://default"/>   
     </transportConnectors>   
    ...   
 </broker>   

            在使用Failover Transport或Discovery transport等能够自动重连的transport的时候，需要注意的是：设想有两个broker，它们都启用AMQ Message Store作为持久化存储，有一个producer和一个consumer连接到某个queue。当因其中一个broker失效时而切换到另一个broker的时候，如果失效的broker的queue中还有未被consumer消费的消息，那么这个queue里的消息仍然滞留在失效broker的中，直到失效的broker被修复并重新切换回这个被修复的broker后，之前被保留的消息才会被consumer消费掉。如果被处理的消息有时序限制，那么应用程序就需要处理这个问题。另外也可以通过ActiveMQ集群来解决这个问题。
        
   在transport重连的时候，可以在connection上注册TransportListener来获得回调，例如：

         Java代码 
          
        
(ActiveMQConnection)connection).addTransportListener(new TransportListener() {   
     public void onCommand(Object cmd) {    
    }   
  
     public void onException(IOException exp) {    
    }   
  
     public void transportInterupted() {    
         // The transport has suffered an interruption from which it hopes to recover.   
    }   
  
     public void transportResumed() {    
         // The transport has resumed after an interruption.    
    }   
});  

关键字: Persistence

2．3 Persistence
2．3．1 AMQ Message Store
AMQ Message Store是ActiveMQ5.0缺省的持久化存储。Message commands被保存到transactional journal（由rolling data logs组成）。Messages被保存到data logs中，同时被reference store进行索引以提高存取速度。Date logs由一些单独的data log文件组成，缺省的文件大小是32M，如果某个消息的大小超过了data log文件的大小，那么可以修改配置以增加data log文件的大小。如果某个data log文件中所有的消息都被成功消费了，那么这个data log文件将会被标记，以便在下一轮的清理中被删除或者归档。以下是其配置的一个例子：

          Xml代码 
           
         
<broker brokerName="broker" persistent="true" useShutdownHook="false">  
   <persistenceAdapter>   
      <amqPersistenceAdapter directory="${activemq.base}/data" maxFileLength="32mb"/>   
   </persistenceAdapter>   
 </broker>   

 
             Property name 
             Default value 
             Comments 
            
             directory 
             activemq-data 
             the path to the directory to use to store the message store data and log files 
            
             useNIO 
             true 
             use NIO to write messages to the data logs 
            
             syncOnWrite 
             false 
             sync every write to disk 
            
             maxFileLength 
             32mb 
             a hint to set the maximum size of the message data logs 
            
             persistentIndex 
             true 
             use a persistent index for the message logs. If this is false, an in-memory structure is maintained 
            
             maxCheckpointMessageAddSize 
             4kb 
             the maximum number of messages to keep in a transaction before automatically committing 
            
             cleanupInterval 
             30000 
             time (ms) before checking for a discarding/moving message data logs that are no longer used 
            
             indexBinSize 
             1024 
             default number of bins used by the index. The bigger the bin size - the better the relative performance of the index 
            
             indexKeySize 
             96 
             the size of the index key - the key is the message id 
            
             indexPageSize 
             16kb 
             the size of the index page - the bigger the page - the better the write performance of the index 
            
             directoryArchive 
             archive 
             the path to the directory to use to store discarded data logs 
            
             archiveDataLogs 
             false 
             if true data logs are moved to the archive directory instead of being deleted

2．3．2 Kaha Persistence
Kaha Persistence 是一个专门针对消息持久化的解决方案。它对典型的消息使用模式进行了优化。在Kaha中，数据被追加到data logs中。当不再需要log文件中的数据的时候，log文件会被丢弃。以下是其配置的一个例子：

          Xml代码 
           
         
<broker brokerName="broker" persistent="true" useShutdownHook="false">  
       <persistenceAdapter>   
         <kahaPersistenceAdapter directory="activemq-data" maxDataFileLength="33554432"/>   
       </persistenceAdapter>   
 </broker>   

2．3．3 JDBC Persistence
目前支持的数据库有Apache Derby, Axion, DB2, HSQL, Informix, MaxDB, MySQL, Oracle, Postgresql, SQLServer, Sybase。
如果你使用的数据库不被支持，那么可以调整StatementProvider 来保证使用正确的SQL方言（flavour of SQL）。通常绝大多数数据库支持以下adaptor：

org.activemq.store.jdbc.adapter.BlobJDBCAdapter
org.activemq.store.jdbc.adapter.BytesJDBCAdapter
org.activemq.store.jdbc.adapter.DefaultJDBCAdapter
org.activemq.store.jdbc.adapter.ImageJDBCAdapter

也可以在配置文件中直接指定JDBC adaptor，例如：

          Xml代码 
           
<jdbcPersistenceAdapter  adapterClass="org.apache.activemq.store.jdbc.adapter.ImageBasedJDBCAdaptor"/>

              以下是其配置的一个例子：
         

          Xml代码 
           
         
<persistence>  
       <jdbcPersistence dataSourceRef=" mysql-ds"/>   
 </persistence>   
  
 <bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">   
       <property name="driverClassName" value="com.mysql.jdbc.Driver"/>   
       <property name="url" value="jdbc:mysql://localhost/activemq?relaxAutoCommit=true"/>   
       <property name="username" value="activemq"/>   
       <property name="password" value="activemq"/>   
       <property name="poolPreparedStatements" value="true"/>   
 </bean>   

              需要注意的是，如果使用MySQL，那么需要设置relaxAutoCommit 标志为true。
         

2．3．4 Disable Persistence
以下是其配置的一个例子：

          Xml代码 
           
         

<broker persistent="false">
</broker>

原文链接:[http://wely.iteye.com/blog/2337715 ]

ActiveMQ - JMS,Transport,Persistence

热门文章

最新文章

相关电子书

Option Name	Default Value	Description
Marshal	false	If true, forces each command sent over the transport to be marshlled and unmarshlled using a WireFormat
wireFormat	default	The name of the WireFormat to use
wireFormat.*		All the properties with this prefix are used to configure the wireFormat
create	true	If the broker should be created on demand if it does not allready exist. Only supported in ActiveMQ 4.1
broker.*		All the properties with this prefix are used to configure the broker. See Configuring Wire Formats for more information

Option Name	Default Value	Description
initialReconnectDelay	10	How long to wait before the first reconnect attempt (in ms)
maxReconnectDelay	30000	The maximum amount of time we ever wait between reconnect attempts (in ms)
useExponentialBackOff	true	Should an exponential backoff be used between reconnect attempts
backOffMultiplier	2	The exponent used in the exponential backoff attempts
maxReconnectAttempts	0	If not 0, then this is the maximum number of reconnect attempts before an error is sent back to the client
randomize	true	use a random algorithm to choose the URI to use for reconnect from the list provided
backup	false	initialize and hold a second transport connection - to enable fast failover

Property name	Default value	Comments
directory	activemq-data	the path to the directory to use to store the message store data and log files
useNIO	true	use NIO to write messages to the data logs
syncOnWrite	false	sync every write to disk
maxFileLength	32mb	a hint to set the maximum size of the message data logs
persistentIndex	true	use a persistent index for the message logs. If this is false, an in-memory structure is maintained
maxCheckpointMessageAddSize	4kb	the maximum number of messages to keep in a transaction before automatically committing
cleanupInterval	30000	time (ms) before checking for a discarding/moving message data logs that are no longer used
indexBinSize	1024	default number of bins used by the index. The bigger the bin size - the better the relative performance of the index
indexKeySize	96	the size of the index key - the key is the message id
indexPageSize	16kb	the size of the index page - the bigger the page - the better the write performance of the index
directoryArchive	archive	the path to the directory to use to store discarded data logs
archiveDataLogs	false	if true data logs are moved to the archive directory instead of being deleted