ActiveMQ - Clustering,Features,Wildcards

关键字: Clustering

2．5 Clustering
    ActiveMQ从多种不同的方面提供了集群的支持。
2．5．1 Queue consumer clusters
    ActiveMQ支持订阅同一个queue的consumers上的集群。如果一个consumer失效，那么所有未被确认（unacknowledged）的消息都会被发送到这个queue上其它的consumers。如果某个consumer的处理速度比其它consumers更快，那么这个consumer就会消费更多的消息。
    需要注意的是，笔者发现AcitveMQ5.0版本的Queue consumer clusters存在一个bug：采用AMQ Message Store，运行一个producer，两个consumer，并采用如下的配置文件：

<beans>
  <broker xmlns="http://activemq.org/config/1.0" brokerName="BugBroker1" useJmx="true">

    <transportConnectors>
      <transportConnector uri="tcp://localhost:61616"/>
    </transportConnectors>

    <persistenceAdapter>
      <amqPersistenceAdapter directory="activemq-data/BugBroker1" maxFileLength="32mb"/>
    </persistenceAdapter>

  </broker>
</beans>

   那么经过一段时间后可能会报出如下错误：

ERROR [ActiveMQ Transport: tcp:///127.0.0.1:1843 - RecoveryListenerAdapter.java:58 - RecoveryListenerAdapter] Message id ID:versus-1837-1203915536609-0:2:1:1:419 could not be recovered from the data store!
    Apache官方文档说，此bug已经被修正，预定在5.1.0版本上体现。

 

2．5．2 Broker clusters
    一个常见的场景是有多个JMS broker，有一个客户连接到其中一个broker。如果这个broker失效，那么客户会自动重新连接到其它的broker。在ActiveMQ中使用failover:// 协议来实现这个功能。ActiveMQ3.x版本的reliable://协议已经变更为failover://。
    如果某个网络上有多个brokers而且客户使用静态发现（使用Static Transport或Failover Transport）或动态发现（使用Discovery Transport），那么客户可以容易地在某个broker失效的情况下切换到其它的brokers。然而，stand alone brokers并不了解其它brokers上的consumers，也就是说如果某个broker上没有consumers，那么这个broker上的消息可能会因得不到处理而积压起来。目前的解决方案是使用Network of brokers，以便在broker之间存储转发消息。ActiveMQ在未来会有更好的特性，用来在客户端处理这个问题。
    从ActiveMQ1.1版本起，ActiveMQ支持networks of brokers。它支持分布式的queues和topics。一个broker会相同对待所有的订阅（subscription）：不管他们是来自本地的客户连接，还是来自远程broker，它都会递送有关的消息拷贝到每个订阅。远程broker得到这个消息拷贝后，会依次把它递送到其内部的本地连接上。有两种方式配置Network of brokers，一种是使用static transport，如下：

<broker brokerName="receiver" persistent="false" useJmx="false">
  <transportConnectors>
    <transportConnector uri="tcp://localhost:62002"/>
  </transportConnectors>
  <networkConnectors>
    <networkConnector uri="static:( tcp://localhost:61616,tcp://remotehost:61616)"/>
  </networkConnectors>
  …
</broker>

    另外一种是使用multicast discovery，如下：

<broker name="sender" persistent="false" useJmx="false">
  <transportConnectors>
    <transportConnector uri="tcp://localhost:0" discoveryUri="multicast://default"/>
  </transportConnectors>
  <networkConnectors>
    <networkConnector uri="multicast://default"/>
  </networkConnectors>
  ...
</broker>

    Network Connector有以下属性：

Property	Default Value	Description
name	bridge	name of the network - for more than one network connector between the same two brokers - use different names
dynamicOnly	false	if true, only forward messages if a consumer is active on the connected broker
decreaseNetworkConsumerPriority	false	decrease the priority for dispatching to a Queue consumer the further away it is (in network hops) from the producer
networkTTL	1	the number of brokers in the network that messages and subscriptions can pass through
conduitSubscriptions	true	multiple consumers subscribing to the same destination are treated as one consumer by the network
excludedDestinations	empty	destinations matching this list won't be forwarded across the network
dynamicallyIncludedDestinations	empty	destinations that match this list will be forwarded across the network n.b. an empty list means all destinations not in the excluded list will be forwarded
staticallyIncludedDestinations	empty	destinations that match will always be passed across the network - even if no consumers have ever registered an interest
duplex	false	if true, a network connection will be used to both produce AND Consume messages. This is useful for hub and spoke scenarios when the hub is behind a firewall etc.

    关于conduitSubscriptions属性，这里稍稍说明一下。设想有两个brokers，分别是brokerA和brokerB，它们之间用forwarding bridge连接。有一个consumer连接到brokerA并订阅queue：Q.TEST。有两个consumers连接到brokerB，也是订阅queue：Q.TEST。这三个consumers有相同的优先级。然后启动一个producer，它发送了30条消息到brokerA。如果conduitSubscriptions=true，那么brokerA上的consumer会得到15条消息， 另外15条消息会发送给brokerB。此时负载并不均衡，因为此时brokerA将brokerB上的两个consumers视为一个；如果conduitSubscriptions=false，那么每个consumer上都会收到10条消息。以下是关于NetworkConnector属性的一个例子：

<networkConnectors>
  <networkConnector uri="static://(tcp://localhost:61617)"
     name="bridge" dynamicOnly="false" conduitSubscriptions="true"
     decreaseNetworkConsumerPriority="false">
     <excludedDestinations>
       <queue physicalName="exclude.test.foo"/>
       <topic physicalName="exclude.test.bar"/>
     </excludedDestinations>
     <dynamicallyIncludedDestinations>
       <queue physicalName="include.test.foo"/>
       <topic physicalName="include.test.bar"/>
     </dynamicallyIncludedDestinations>
     <staticallyIncludedDestinations>
       <queue physicalName="always.include.queue"/>
       <topic physicalName="always.include.topic"/>
     </staticallyIncludedDestinations>
  </networkConnector>
</networkConnectors>

 

2．5．3 Master Slave
    在一个网络内运行多个brokers或者stand alone brokers时存在一个问题，这就是消息在物理上只被一个broker持有，因此当某个broker失效，那么你只能等待直到它重启后，这个broker上的消息才能够被继续发送（如果没有设置持久化，那么在这种情况下，消息将会丢失）。Master Slave 背后的想法是，消息被复制到slave broker，因此即使master broker遇到了像硬件故障之类的错误，你也可以立即切换到slave broker而不丢失任何消息。
    Master Slave是目前ActiveMQ推荐的高可靠性和容错的解决方案。以下是几种不同的类型：

Master Slave Type	Requirements	Pros	Cons
Pure Master Slave	None	No central point of failure	Requires manual restart to bring back a failed master and can only support 1 slave
Shared File System Master Slave	A Shared File system such as a SAN	Run as many slaves as required. Automatic recovery of old masters	Requires shared file system
JDBC Master Slave	A Shared database	Run as many slaves as required. Automatic recovery of old masters	Requires a shared database. Also relatively slow as it cannot use the high performance journal

 

2．5．3．1 Pure Master Slave
    Pure Master Slave的工作方式如下：

• Slave broker消费master broker上所有的消息状态，例如消息、确认和事务状态等。只要slave broker连接到了master broker，它不会（也不被允许）启动任何network connectors或者transport connectors，所以唯一的目的就是复制master broker的状态。
• Master broker只有在消息成功被复制到slave broker之后才会响应客户。例如，客户的commit请求只有在master broker和slave broker都处理完毕commit请求之后才会结束。
• 当master broker失效的时候，slave broker有两种选择，一种是slave broker启动所有的network connectors和transport connectors，这允许客户端切换到slave broker；另外一种是slave broker停止。这种情况下，slave broker只是复制了master broker的状态。
• 客户应该使用failover transport并且应该首先尝试连接master broker。例如：
  failover://(tcp://masterhost:61616,tcp://slavehost:61616)?randomize=false
  设置randomize为false就可以让客户总是首先尝试连接master broker（slave broker并不会接受任何连接，直到它成为了master broker）。

   Pure Master Slave具有以下限制：

• 只能有一个slave broker连接到master broker。
• 在因master broker失效而导致slave broker成为master之后，之前的master broker只有在当前的master broker（原slave broker）停止后才能重新生效。
• Master broker失效后而切换到slave broker后，最安全的恢复master broker的方式是人工处理。首先要停止slave broker（这意味着所有的客户也要停止）。然后把slave broker的数据目录中所有的数据拷贝到master broker的数据目录中。然后重启master broker和slave broker。

   Master broker不需要特殊的配置。Slave broker需要进行以下配置

<broker masterConnectorURI="tcp://masterhost:62001" shutdownOnMasterFailure="false">
    ...
    <transportConnectors>
      <transportConnector uri="tcp://slavehost:61616"/>
   </transportConnectors>
</broker>

    其中的masterConnectorURI用于指向master broker，shutdownOnMasterFailure用于指定slave broker在master broker失效的时候是否需要停止。此外，也可以使用如下配置：

<broker brokerName="slave" useJmx="false"  deleteAllMessagesOnStartup="true"  xmlns="http://activemq.org/config/1.0">
  ...
  <services>
    <masterConnector remoteURI= "tcp://localhost:62001" userName="user" password="password"/>
  </services>
</broker>

   需要注意的是，笔者认为ActiveMQ5.0版本的Pure Master Slave仍然不够稳定。

 

2．5．3．2 Shared File System Master Slave
    如果你使用SAN或者共享文件系统，那么你可以使用Shared File System Master Slave。基本上，你可以运行多个broker，这些broker共享数据目录。当第一个broker得到文件上的排他锁之后，其它的broker便会在循环中等待获得这把锁。客户端使用failover transport来连接到可用的broker。当master broker失效的时候会释放这把锁，这时候其中一个slave broker会得到这把锁从而成为master broker。以下是ActiveMQ配置的一个例子：

<broker useJmx="false"  xmlns="http://activemq.org/config/1.0">
   <persistenceAdapter>
      <journaledJDBC dataDirectory="/sharedFileSystem/broker"/>
   </persistenceAdapter>
   …
</broker>

 

2．5．3．3 JDBC Master Slave
    JDBC Master Slave的工作原理跟Shared File System Master Slave类似，只是采用了数据库作为持久化存储。以下是ActiveMQ配置的一个例子：

<beans>
  <broker xmlns="http://activemq.org/config/1.0" brokerName="JdbcMasterBroker">
    ...
    <persistenceAdapter>
      <jdbcPersistenceAdapter dataSource="#mysql-ds"/>
    </persistenceAdapter>

  </broker>

  <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:3306/test?relaxAutoCommit=true"/>
    <property name="username" value="username"/>
    <property name="password" value="passward"/>
    <property name="poolPreparedStatements" value="true"/>
  </bean>
</beans>

   需要注意的是，如果你使用MySQL数据库，需要首先执行以下三条语句：（Apache官方文档说，此bug已经被修正，预定在5.1.0版本上体现） 

Sql代码 

ALTER TABLE activemq_acks ENGINE = InnoDB;
ALTER TABLE activemq_lock ENGINE = InnoDB;
ALTER TABLE activemq_msgs ENGINE = InnoDB;

 

关键字: Features

2．6 Features
    ActiveMQ包含了很多功能强大的特性，下面简要介绍其中的几个。
2．6．1 Exclusive Consumer
    Queue中的消息是按照顺序被分发到consumers的。然而，当你有多个consumers同时从相同的queue中提取消息时，你将失去这个保证。因为这些消息是被多个线程并发的处理。有的时候，保证消息按照顺序处理是很重要的。例如，你可能不希望在插入订单操作结束之前执行更新这个订单的操作。
    ActiveMQ从4.x版本起开始支持Exclusive Consumer （或者说Exclusive Queues）。 Broker会从多个consumers中挑选一个consumer来处理queue中所有的消息，从而保证了消息的有序处理。如果这个consumer失效，那么broker会自动切换到其它的consumer。
    可以通过Destination Options 来创建一个Exclusive Consumer，如下：

queue = new ActiveMQQueue("TEST.QUEUE?consumer.exclusive=true");
consumer = session.createConsumer(queue);

    顺便说一下，可以给consumer设置优先级，以便针对网络情况（如network hops）进行优化，如下：

Java代码  

queue = new ActiveMQQueue("TEST.QUEUE?consumer.exclusive=true &consumer.priority=10");

 

2．6．2 Message Groups
    用Apache官方文档的话说，Message Groups rock！它是Exclusive Consumer功能的增强。逻辑上，Message Groups 可以看成是一种并发的Exclusive Consumer。跟所有的消息都由唯一的consumer处理不同，JMS 消息属性JMSXGroupID 被用来区分message group。Message Groups特性保证所有具有相同JMSXGroupID 的消息会被分发到相同的consumer（只要这个consumer保持active）。另外一方面，Message Groups特性也是一种负载均衡的机制。
    在一个消息被分发到consumer之前，broker首先检查消息JMSXGroupID属性。如果存在，那么broker 会检查是否有某个consumer拥有这个message group。如果没有，那么broker会选择一个consumer，并将它关联到这个message group。此后，这个consumer会接收这个message group的所有消息，直到：

• Consumer被关闭。
• Message group被关闭。通过发送一个消息，并设置这个消息的JMSXGroupSeq为0。

   从4.1版本开始，ActiveMQ支持一个布尔字段JMSXGroupFirstForConsumer 。当某个message group的第一个消息被发送到consumer的时候，这个字段被设置。如果客户使用failover transport连接到broker。在由于网络问题等造成客户重新连接到broker的时候，相同message group的消息可能会被分发到不同与之前的consumer，因此JMSXGroupFirstForConsumer字段也会被重新设置。 

   以下是使用message groups的例子：

Java代码  

Mesasge message = session.createTextMessage("<foo>hey</foo>");
message.setStringProperty("JMSXGroupID", "IBM_NASDAQ_20/4/05");
...
producer.send(message);

2．6．3 JMS Selectors
    JMS Selectors用于在订阅中，基于消息属性对进行消息的过滤。JMS Selectors由SQL92语法定义。以下是个Selectors的例子：

Java代码  

consumer = session.createConsumer(destination, "JMSType = 'car' AND weight > 2500");

     在JMS Selectors表达式中，可以使用IN、NOT IN、LIKE等，例如：
    LIKE '12%3' （'123' true，'12993' true，'1234' false）
    LIKE 'l_se' （'lose' true，'loose' false）
    LIKE '\_%' ESCAPE '\' （'_foo' true，'foo' false）
    需要注意的是，JMS Selectors表达式中的日期和时间需要使用标准的long型毫秒值。另外表达式中的属性不会自动进行类型转换，例如：

Java代码  

myMessage.setStringProperty("NumberOfOrders", "2");

    "NumberOfOrders > 1" 求值结果是false。关于JMS Selectors的详细文档请参考javax.jms.Message的javadoc。
    上一小节介绍的Message Groups虽然可以保证具有相同message group的消息被唯一的consumer顺序处理，但是却不能确定被哪个consumer处理。在某些情况下，Message Groups可以和JMS Selector一起工作，例如：
    设想有三个consumers分别是A、B和C。你可以在producer中为消息设置三个message groups分别是"A"、"B"和"C"。然后令consumer A使用"JMXGroupID = 'A'"作为selector。B和C也同理。这样就可以保证message group A的消息只被consumer A处理。需要注意的是，这种做法有以下缺点：

• producer必须知道当前正在运行的consumers，也就是说producer和consumer被耦合到一起。
• 如果某个consumer失效，那么应该被这个consumer消费的消息将会一直被积压在broker上。

2．6．4 Pending Message Limit Strategy
    首先简要介绍一下prefetch机制。ActiveMQ通过prefetch机制来提高性能，这意味这客户端的内存里可能会缓存一定数量的消息。缓存消息的数量由prefetch limit来控制。当某个consumer的prefetch buffer已经达到上限，那么broker不会再向consumer分发消息，直到consumer向broker发送消息的确认。可以通过在ActiveMQConnectionFactory或者ActiveMQConnection上设置ActiveMQPrefetchPolicy对象来配置prefetch policy。也可以通过connection options或者destination options来配置。例如：
    tcp://localhost:61616?jms.prefetchPolicy.all=50
    tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=1
    queue = new ActiveMQQueue("TEST.QUEUE?consumer.prefetchSize=10");
    prefetch size的缺省值如下：

• persistent queues (default value: 1000)
• non-persistent queues (default value: 1000)
• persistent topics (default value: 100)
• non-persistent topics (default value: Short.MAX_VALUE -1)

    慢消费者会在非持久的topics上导致问题：一旦消息积压起来，会导致broker把大量消息保存在内存中，broker也会因此而变慢。未来ActiveMQ可能会实现磁盘缓存，但是这也还是会存在性能问题。目前ActiveMQ使用Pending Message Limit Strategy来解决这个问题。除了prefetch buffer之外，你还要配置缓存消息的上限，超过这个上限后，新消息到来时会丢弃旧消息。通过在配置文件的destination map中配置PendingMessageLimitStrategy，可以为不用的topic namespace配置不同的策略。目前有以下两种：

• ConstantPendingMessageLimitStrategy。这个策略使用常量限制。
  例如：<constantPendingMessageLimitStrategy limit="50"/>
• PrefetchRatePendingMessageLimitStrategy。这个策略使用prefetch size的倍数限制。
  例如：<prefetchRatePendingMessageLimitStrategy multiplier="2.5"/>

   在以上两种方式中，如果设置0意味着除了prefetch之外不再缓存消息；如果设置-1意味着禁止丢弃消息。  
    此外，你还可以配置消息的丢弃策略，目前有以下两种：

• oldestMessageEvictionStrategy。这个策略丢弃最旧的消息。
• oldestMessageWithLowestPriorityEvictionStrategy。这个策略丢弃最旧的，而且具有最低优先级的消息。

   以下是个ActiveMQ配置文件的例子：

Xml代码  

<broker persistent="false" brokerName="${brokername}" xmlns="http://activemq.org/config/1.0">
    <destinationPolicy>
      <policyMap>
        <policyEntries>
          <policyEntry topic="PRICES.>">
            <!--  10 seconds worth -->
            <subscriptionRecoveryPolicy>
              <timedSubscriptionRecoveryPolicy recoverDuration="10000" />
            </subscriptionRecoveryPolicy>

            <!-- lets force old messages to be discarded for slow consumers -->
            <pendingMessageLimitStrategy>
              <constantPendingMessageLimitStrategy limit="10"/>
            </pendingMessageLimitStrategy>
          </policyEntry>
        </policyEntries>
      </policyMap>
    </destinationPolicy>
    ...
</broker>

 

2．6．5 Composite Destinations
    从1.1版本起, ActiveMQ支持composite destinations。它允许用一个虚拟的destination 代表多个destinations。例如你可以通过composite destinations在一个操作中同时向12个queue发送消息。在composite destinations中，多个destination之间采用","分割。例如：

Java代码  

Queue queue = new ActiveMQQueue("FOO.A,FOO.B,FOO.C");

   如果你希望使用不同类型的destination，那么需要加上前缀如queue:// 或topic://，例如： 

Java代码  

Queue queue = new ActiveMQQueue("FOO.A,topic://NOTIFY.FOO.A");

   以下是ActiveMQ配置文件进行配置的一个例子：

Xml代码  

<destinationInterceptors>
  <virtualDestinationInterceptor>
    <virtualDestinations>
      <compositeQueue name="MY.QUEUE">
        <forwardTo>
          <queue physicalName="FOO" />
          <topic physicalName="BAR" />
        </forwardTo>
      </compositeQueue>
    </virtualDestinations>
  </virtualDestinationInterceptor>
</destinationInterceptors>

   可以在转发前，先通过JMS Selector判断一个消息是否需要转发，例如：

Xml代码  

<destinationInterceptors>
  <virtualDestinationInterceptor>
    <virtualDestinations>
      <compositeQueue name="MY.QUEUE">
        <forwardTo>
          <filteredDestination selector="odd = 'yes'" queue="FOO"/>
          <filteredDestination selector="i = 5" topic="BAR"/>
        </forwardTo>
      </compositeQueue>
    </virtualDestinations>
  </virtualDestinationInterceptor>
</destinationInterceptors>

 

2．6．6 Mirrored Queues
    每个queue中的消息只能被一个consumer消费。然而，有时候你可能希望能够监视生产者和消费者之间的消息流。你可以通过使用Virtual Destinations 来建立一个virtual queue 来把消息转发到多个queues中。但是 为系统中每个queue都进行如此的配置可能会很麻烦。 
    ActiveMQ支持Mirrored Queues。Broker会把发送到某个queue的所有消息转发到一个名称类似的topic，因此监控程序可以订阅这个mirrored queue topic。为了启用Mirrored Queues，首先要将BrokerService的useMirroredQueues属性设置成true，然后可以通过destinationInterceptors设置其它属性，如mirror topic的前缀，缺省是"VirtualTopic.Mirror."。以下是ActiveMQ配置文件的一个例子：

Xml代码  

<broker xmlns="http://activemq.org/config/1.0" brokerName="MirroredQueuesBroker1" useMirroredQueues="true">

  <transportConnectors>
    <transportConnector uri="tcp://localhost:61616"/>
  </transportConnectors>

  <destinationInterceptors>
      <mirroredQueue copyMessage = "true" prefix="Mirror.Topic"/>
  </destinationInterceptors>
  ...
</broker>

    假如某个producer向名为Foo.Bar的queue中发送消息，那么你可以通过订阅名为Mirror.Topic.Foo.Bar的topic来获得发送到Foo.Bar中的所有消息。 

关键字: Wildcards

2．6．7 Wildcards
    Wildcards用来支持联合的名字分层体系（federated name hierarchies）。它不是JMS规范的一部分，而是ActiveMQ的扩展。ActiveMQ支持以下三种wildcards：

• "." 用于作为路径上名字间的分隔符。
• "*" 用于匹配路径上的任何名字。
• ">" 用于递归地匹配任何以这个名字开始的destination。

   作为一种组织事件和订阅感兴趣那部分信息的一种方法，这个概念在金融市场领域已经流行了一段时间了。设想你有以下两个destination：

• PRICE.STOCK.NASDAQ.IBM （IBM在NASDAQ的股价）
• PRICE.STOCK.NYSE.SUNW （SUN在纽约证券交易所的股价）

   订阅者可以明确地指定destination的名字来订阅消息，或者它也可以使用wildcards来定义一个分层的模式来匹配它希望订阅的destination。例如：

Subscription	Meaning
PRICE.>	Any price for any product on any exchange
PRICE.STOCK.>	Any price for a stock on any exchange
PRICE.STOCK.NASDAQ.*	Any stock price on NASDAQ
PRICE.STOCK.*.IBM	Any IBM stock price on any exchange

 

2．6．8 Async Sends
    ActiveMQ支持以同步（sync）方式或者异步（async）方式向broker发送消息。 使用何种方式对send方法的延迟有巨大的影响。对于生产者来说，既然延迟是决定吞吐量的重要因素，那么使用异步发送方式会极大地提高系统的性能。
    ActiveMQ缺省使用异步传输方式。但是按照JMS规范，当在事务外发送持久化消息的时候，ActiveMQ会强制使用同步发送方式。在这种情况下，每一次发送都是同步的，而且阻塞到收到broker的应答。这个应答保证了broker已经成功地将消息持久化，而且不会丢失。但是这样作也严重地影响了性能。
    如果你的系统可以容忍少量的消息丢失，那么可以在事务外发送持久消息的时候，选择使用异步方式。以下是几种不同的配置方式：

Java代码  

cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");
((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);
((ActiveMQConnection)connection).setUseAsyncSend(true);

 

2．6．9 Dispatch Policies
2．6．9．1 Round Robin Dispatch Policy
    在2．6．4小节介绍过ActiveMQ的prefetch机制，ActiveMQ的缺省参数是针对处理大量消息时的高性能和高吞吐量而设置的。所以缺省的prefetch参数比较大，而且缺省的dispatch policies会尝试尽可能快的填满prefetch缓冲。然而在有些情况下，例如只有少量的消息而且单个消息的处理时间比较长，那么在缺省的prefetch和dispatch policies下，这些少量的消息总是倾向于被分发到个别的consumer上。这样就会因为负载的不均衡分配而导致处理时间的增加。
    Round robin dispatch policy会尝试平均分发消息，以下是ActiveMQ配置文件的一个例子：

Xml代码  

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry topic="FOO.>">
        <dispatchPolicy>
          <roundRobinDispatchPolicy />
        </dispatchPolicy>
      </policyEntry>
    </policyEntries>
  </policyMap>
</destinationPolicy>

 

2．6．9．2 Strict Order Dispatch Policy
    有时候需要保证不同的topic consumer以相同的顺序接收消息。通常ActiveMQ会保证topic consumer以相同的顺序接收来自同一个producer的消息。然而，由于多线程和异步处理，不同的topic consumer可能会以不同的顺序接收来自不同producer的消息。例如有两个producer，分别是P和Q。差不多是同一时间内，P发送了P1、P2和P3三个消息；Q发送了Q1和Q2两个消息。两个不同的consumer可能会以以下顺序接收到消息：

   consumer1: P1 P2 Q1 P3 Q2
    consumer2: P1 Q1 Q2 P2 P3
    Strict order dispatch policy 会保证每个topic consumer会以相同的顺序接收消息，代价是性能上的损失。以下是采用了strict order dispatch policy后，两个不同的consumer可能以以下的顺序接收消息：
    consumer1: P1 P2 Q1 P3 Q2
    consumer2: P1 P2 Q1 P3 Q2

   以下是ActiveMQ配置文件的一个例子：

Xml代码  

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry topic=""FOO.>">
        <dispatchPolicy>
          <strictOrderDispatchPolicy />
        </dispatchPolicy>
      </policyEntry>
    </policyEntries>
  </policyMap>
</destinationPolicy>

 

2．6．10 Message Cursors
    当producer发送的持久化消息到达broker之后，broker首先会把它保存在持久存储中。接下来，如果发现当前有活跃的consumer，而且这个consumer消费消息的速度能跟上producer生产消息的速度，那么ActiveMQ会直接把消息传递给broker内部跟这个consumer关联的dispatch queue；如果当前没有活跃的consumer或者consumer消费消息的速度跟不上producer生产消息的速度，那么ActiveMQ会使用Pending Message Cursors保存对消息的引用。在需要的时候，Pending Message Cursors把消息引用传递给broker内部跟这个consumer关联的dispatch queue。以下是两种Pending Message Cursors：

• VM Cursor。在内存中保存消息的引用。
• File Cursor。首先在内存中保存消息的引用，如果内存使用量达到上限，那么会把消息引用保存到临时文件中。

   在缺省情况下，ActiveMQ 5.0根据使用的Message Store来决定使用何种类型的Message Cursors，但是你可以根据destination来配置Message Cursors。

    对于topic，可以使用的pendingSubscriberPolicy 有vmCursor和fileCursor。可以使用的PendingDurableSubscriberMessageStoragePolicy有vmDurableCursor 和 fileDurableSubscriberCursor。以下是ActiveMQ配置文件的一个例子：

Xml代码  

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry topic="org.apache.>">
        <pendingSubscriberPolicy>
          <vmCursor />
        </pendingSubscriberPolicy>
        <PendingDurableSubscriberMessageStoragePolicy>
          <vmDurableCursor/>
        </PendingDurableSubscriberMessageStoragePolicy>
      </policyEntry>
    </policyEntries>
  </policyMap>
</destinationPolicy>

   对于queue，可以使用的pendingQueuePolicy有vmQueueCursor 和 fileQueueCursor。以下是ActiveMQ配置文件的一个例子：

Xml代码  

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry queue="org.apache.>">
        <pendingQueuePolicy>
          <vmQueueCursor />
        </pendingQueuePolicy>
      </policyEntry>
    </policyEntries>
  </policyMap>
</destinationPolicy>

 

2．6．11 Optimized Acknowledgement
    ActiveMQ缺省支持批量确认消息。由于批量确认会提高性能，因此这是缺省的确认方式。如果希望在应用程序中禁止经过优化的确认方式，那么可以采用如下方法：

Java代码  

cf = new ActiveMQConnectionFactory ("tcp://locahost:61616?jms.optimizeAcknowledge=false");
((ActiveMQConnectionFactory)connectionFactory).setOptimizeAcknowledge(false);
((ActiveMQConnection)connection).setOptimizeAcknowledge(false);

 

2．6．12 Producer Flow Control
    同步发送消息的producer会自动使用producer flow control ；对于异步发送消息的producer，要使用producer flow control，你先要为connection配置一个ProducerWindowSize参数，如下：

Java代码  

((ActiveMQConnectionFactory)cf).setProducerWindowSize(1024000);

    ProducerWindowSize是producer在发送消息的过程中，收到broker对于之前发送消息的确认之前， 能够发送消息的最大字节数。你也可以禁用producer flow control，以下是ActiveMQ配置文件的一个例子：

Java代码  

<destinationPolicy>
  <policyMap>
    <policyEntries>
      <policyEntry topic="FOO.>" producerFlowControl="false">
        <dispatchPolicy>
          <strictOrderDispatchPolicy/>
        </dispatchPolicy>
      </policyEntry>
    </policyEntries>
  </policyMap>
</destinationPolicy>

 

2．6．13 Message Transformation
    有时候需要在JMS provider内部进行message的转换。从4.2版本起，ActiveMQ 提供了一个MessageTransformer 接口用于进行消息转换，如下：

Java代码  

public interface MessageTransformer {
    Message producerTransform(Session session, MessageProducer producer, Message message) throws JMSException;
    Message consumerTransform(Session session, MessageConsumer consumer, Message message)throws JMSException;
}

    通过在以下对象上通过调用setTransformer方法来设置MessageTransformer：

• ActiveMQConnectionFactory
• ActiveMQConnection
• ActiveMQSession
• ActiveMQMessageConsumer
• ActiveMQMessageProducer

   MessageTransformer接口支持：

• 在消息被发送到JMS provider的消息总线前进行转换。通过producerTransform方法。
• 在消息到达消息总线后，但是在consumer接收到消息前进行转换。通过consumerTransform方法。

   以下是个简单的例子：  

 

Java代码  

public class SimpleMessage implements Serializable {
    //
    private static final long serialVersionUID = 2251041841871975105L;

    //
    private String id;
    private String text;

    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getText() {
        return text;
    }
    public void setText(String text) {
        this.text = text;
    }
}

    在producer内发送ObjectMessage，如下：

Java代码  

SimpleMessage sm = new SimpleMessage();
sm.setId("1");
sm.setText("this is a sample message");
ObjectMessage message = session.createObjectMessage();
message.setObject(sm);
producer.send(message);

   在consumer的session上设置一个MessageTransformer用于将ObjectMessage转换成TextMessage，如下：

((ActiveMQSession)session).setTransformer(new MessageTransformer() {
public Message consumerTransform(Session session, MessageConsumer consumer, Message message) throws JMSException {
    ObjectMessage om = (ObjectMessage)message;
    XStream xstream = new XStream();
    xstream.alias("simple message", SimpleMessage.class);
    String xml = xstream.toXML(om.getObject());
    return session.createTextMessage(xml);
}

public Message producerTransform(Session session, MessageProducer consumer, Message message) throws JMSException {
    return null;
}
});