ActiveMQ消息策略

素颜马尾好姑娘i 2022-06-13 08:25 308阅读 0赞

**ActiveMQ中提供了众多的“策略”(policy)**,它们可以在broker端为每个通道“定制”消息的管理方式。本文将简单描述主要的几种Policy。

**一. DispatchPolcicy: 转发策略(Topic)**

此策略表明broker端消息转发给多个Consumer时，消息被发送的顺序性，这个顺序通常指Consumer的顺序，只对Topic有效，它有3种常用的类型：

*1) RoundRobinDispatchPolicy:* “轮询”，消息将依次发送给每个“订阅者”。“订阅者”列表默认按照订阅的先后顺序排列，在转发消息时，对于匹配消息的第一个订阅者，将会被移动到“订阅者”列表的尾部，这也意味着“下一条”消息，将会较晚的转发给它。

*2) StrictOrderDispatchPolicy:* 严格有序，消息依次发送给每个订阅者，按照“订阅者”订阅的时间先后。它和RoundRobin最大的区别是，没有移动“订阅者”顺序的操作。

*3) PriorityDispatchPolicy:* 基于“property”权重对“订阅者”排序。它要求开发者首先需要对每个订阅者指定priority，默认每个consumer的权重都一样。

*4) SimpleDispatchPolicy:* 默认值，按照当前“订阅者”列表的顺序。其中PriorityDispatchPolicy是其子类。

"轮询"是比较常用的策略。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry topic=">">                 
2.    <dispatchPolicy>    
3.     <roundRobinDispatchPolicy/>  
4.    </dispatchPolicy>  
5.  </policyEntry>

**二.SubscriptionRecoveryPolicy:** 恢复策略(Topic)

在非耐久订阅者“失效”期间或一个新的Topic，broker可以保留的可追溯的消息量。前提是Topic必须是“retroactive”，我们可以在distination地址中指定此属性,例如："order.topic?consumer.retroactive=true"。默认情况下，订阅者只能获取“订阅”开始之后的消息，如果Retroactive=true，那么订阅者就可以获取其创建之前的消息列表。此Policy就是用来控制“retroactive”的消息量。

*1) FixedSizedSubscriptionRecoveryPolicy:* 保存一定size的消息，broker将为此Topic开辟定额的RAM用来保存最新的消息。

Java代码   ![收藏代码][icon_star.png]

1.    
2.  <fixedSizedSubscriptionRecoveryPolicy maximumSize="1024"/>

*2) FixedCountSubscriptionRecoveryPolicy:* 保存一定条数的消息。

Java代码   ![收藏代码][icon_star.png]

1.    
2.  <fixedCountSubscriptionRecoveryPolicy maximumSize="100"/>

*3) LastImageSubscriptionRecoveryPolicy:* 只保留最新的一条数据

*4) QueryBasedSubscriptionRecoveryPolicy:* 符合置顶selector的消息都将被保存，具体能够“恢复”多少消息，由底层存储机制决定；比如对于非持久化消息，只要内存中还存在，则都可以恢复。

*5) TimedSubscriptionRecoveryPolicy:* 保留最近一段时间的消息。

Java代码   ![收藏代码][icon_star.png]

1.    
2.  <timedSubscriptionRecoveryPolicy recoverDuration="60000" />

* 6) NoSubscriptionRecoveryPolicy:* 关闭“恢复机制”。默认值。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry topic=">">                 
2.    <subscriptionRecoveryPolicy>    
3.     <fixedCountSubscriptionRecoveryPolicy maximumSize="100"/>  
4.    </subscriptionRecoveryPolicy>  
5.  </policyEntry

**三. DeadLetterStrategy:** “死信”策略

Broker将如何管理“死信”。当消息过期后，或者“重发”几次之后仍然不能被正常消息，那么此消息将会被移除到DeadLetter队列中。此后，我们可以通过侦听死信队列，来获取相关通知或者对消息做额外的操作。

*1)  IndividualDeadLetterStrategy:* 把DeadLetter放入各自的死信通道中，对于Queue而言，死信通道的前缀默认为“ActiveMQ.DLQ.Queue.”，Topic为“ActiveMQ.DLQ.Topic.”；比如队列Order，那么它对应的死信通道为“ActiveMQ.DLQ.Queue.Order”。我们使用“queuePrefix”“topicPrefix”来指定上述前缀。

默认情况下，无论是Topic还是Queue，broker将使用Queue来保存DeadLeader，即死信通道通常为Queue；不过开发者也可以指定为Topic。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry queue="order">  
2.    <deadLetterStrategy>  
3.      <individualDeadLetterStrategy  
4.        queuePrefix="DLQ." useQueueForQueueMessages="false" />  
5.    </deadLetterStrategy>  
6.  </policyEntry>

上述将队列Order中出现的DeadLetter保存在DLQ.Order中，不过此时DLQ.Order为Topic。individualDeadLetterStrategy还有一个属性为“useQueueForTopicMessages”，此值表示是否将Topic的DeadLetter保存在Queue中，默认为true。

*2) SharedDeadLetterStrategy: *将所有的DeadLetter保存在一个共享的队列中，这是ActiveMQ broker端默认的策略。共享队列默认为“ActiveMQ.DLQ”，可以通过“deadLetterQueue”属性来设定。

Java代码   ![收藏代码][icon_star.png]

1.  <deadLetterStrategy>  
2.      <sharedDeadLetterStrategy deadLetterQueue="DLQ-QUEUE"/>  
3.   </deadLetterStrategy>

*3) DiscardingDeadLetterStrategy:* broker将直接抛弃DeadLeatter。如果开发者不需要关心DeadLetter，可以使用此策略。AcitveMQ提供了一个便捷的插件：DiscardingDLQBrokerPlugin，来抛弃DeadLetter。

Java代码   ![收藏代码][icon_star.png]

1.  <broker>  
2.      <plugins>  
3.        <discardingDLQBrokerPlugin dropAll="true" dropTemporaryTopics="true" dropTemporaryQueues="true" />  
4.      </plugins>  
5.  </broker>

对于上述三种策略，还有2个很重要的可选参数，“processExpired”表示是否将过期消息放入死信队列，默认为true；“processNonPersistent”表示是否将“非持久化”消息放入死信队列，默认为false。

**四. PendingMessageLimitStrategy:** 消息限制策略(面向Slow Consumer)

此策略只对Topic有效，只对nondurable订阅者有效，当通道中有大量的消息积压时，broker可以保留的消息量。为了防止Topic中有慢速消费者，导致整个通道消息积压。(对于Topic而言，一条消息只有所有的订阅者都消费才会被删除)

*1) ConstantPendingMessageLimitStrategy:* 保留固定条数的消息，如果消息量超过limit，将使用“MessageEvictionStrategy”移除消息（参见下文）。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry topic="PRICES.>">  
2.        
3.      <pendingMessageLimitStrategy>  
4.          <constantPendingMessageLimitStrategy limit="50"/>  
5.      </pendingMessageLimitStrategy>  
6.  </policyEntry>

*2) PrefetchRatePendingMessageLimitStrategy:* 保留prefetchSize倍数条消息。

Java代码   ![收藏代码][icon_star.png]

1.    
2.  <prefetchRatePendingMessageLimitStrategy multiplier="2.5"/>

**五. MessageEvictionStrategy:** 消息剔除策略(面向Slow Consumer)

配合PendingMessageLimitStrategy，只对Topic有效,只对nondurable订阅者有效。当PendingMessage的数量超过限制时，broker该如何剔除多余的消息。当Topic接收到信息消息后，会将消息“Copy”给每个订阅者，在保存这个消息时(保存策略参见【八】)，将会检测pendingMessages的数量是否超过限制(由【四】来检测)，如果超过限制，将会在pendingMessages中使用MessageEvicationStrategy移除多余的消息，此后将新消息保存在PendingMessages中。

*1) OldestMessageEvictionStrategy:* 移除旧消息，默认策略。

*2) OldestMessageWithLowestPriorityEvictionStrategy:* 旧数据中权重较低的消息，将会被移除。(message.getPriority())

*3) UniquePropertyMessageEvictionStrategy:* 移除具有指定property的旧消息。开发者可以指定property的名称，从此属性值相同的消息列表中移除较旧的（根据消息的创建时间）。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry topic="SKU.PRICE.>">  
2.        
3.      <pendingMessageLimitStrategy>  
4.          <constantPendingMessageLimitStrategy limit="10000"/>  
5.      </pendingMessageLimitStrategy>  
6.      <messageEvictionStrategy>  
7.          <uniquePropertyMessageEvictionStrategy propertyName="SKU" />  
8.      </messageEvictionStrategy>  
9.  </policyEntry>

上述配置，针对SKU.PRICE通道中，只保留10000个最新的消息，当容量达到阀值时，将SKU值相同的消息列表中较旧的消息移除（只保留最新的一条消息）。比如在每条消息中，都封装一个SKU（商品ID）以及其最新价格，那么通过这种策略，值保留相同SKU的最新的一个消息。

**六. SlowConsumerStrategy:** 慢速消费者策略

Broker将如何处理慢消费者。Broker将会启动一个后台线程用来检测所有的慢速消费者，并定期关闭关闭它们。

*1) AbortSlowConsumerStrategy:* 中断慢速消费者，慢速消费将会被关闭。

Java代码   ![收藏代码][icon_star.png]

1.  <slowConsumerStrategy>    
2.      <abortSlowConsumerStrategy abortConnection="false"/>    
3.  </slowConsumerStrategy>

*2) AbortSlowConsumerStrategy:* 如果慢速消费者最后一个ACK距离现在的时间间隔超过阀值，则中断慢速消费者。

Java代码   ![收藏代码][icon_star.png]

1.  <slowConsumerStrategy>    
2.      <abortSlowConsumerStrategy  maxTimeSinceLastAck="30000"/>    
3.  </slowConsumerStrategy>

**七. PendingQueueMessageStoragePolicy:** 待消费消息策略

当通道中有大量Slow Consumer时，Broker该如何优化消息的转发，以及在此情况下，“非持久化”消息达到内存限制时该如何处理。

当Broker接受到消息后，通常将最新的消息写入内存以提高消息转发的效率，提高消息ACK的效率，减少对对底层Store的操作；如果Consumer非常快速，那么消息将会立即转发给Consumer，不需要额外的操作；但当遇到Slow Consumer时，情况似乎并没有那么美好。

持久化消息，通常为：写入Store-->线程轮询，从Store中pageIn数据到PendingStorage-->转发给Consumer-->从PendingStorage中移除-->消息ACK后从Store中移除。

对于非持久化数据，通常为：写入内存-->如果内存足够，则PendingStorage直接以内存中的消息转发-->如果内存不足，则将内存中的消息swap到临时文件中-->从临时文件中pageIn到内存，转发给Consumer。

AcitveMQ提供了几个的Cursor机制，它就是用来保存Pending Messages。

*1) vmQueueCursor:* 将待转发消息保存在额外的内存(JVM linkeList)的存储结构中。是“非持久化消息”的默认设置，如果Broker不支持Persistent，它是任何类型消息的默认设置。有OOM风险。

*2) fileQueueCursor:* 将消息保存到临时文件中。文件存储方式有broker的tempDataStore属性决定。是“持久化消息”的默认设置。

*3) storeCursor:* “综合”设置，对于非持久化消息，将采用vmQueueCursor存储，对于持久化消息采用fileQueueCursor。这是强烈推荐的策略，也是效率最好的策略。

Java代码   ![收藏代码][icon_star.png]

1.    
2.  <broker persistent="true">  
3.      <persistenceAdapter>  
4.            
7.          <levelDB directory="$\{activemq.data\}/leveldb"/>  
8.      </persistenceAdapter>  
9.        
10.     <tempDataStore>  
11.           
14.         <levelDB directory="$\{activemq.data\}/leveldb/tmp"/>  
15.     </tempDataStore>  
16.       
17.     <policyEntry queue=">" producerFlowControl="true" memoryLimit="512mb">                 
18.         <pendingQueuePolicy>    
19.             <storeCursor>  
20.                 <nonPersistent>  
21.                     <fileQueueCursor/>  
22.                 </nonPersistent>  
23.             </storeCursor>  
24.         </pendingQueuePolicy>    
25.     </policyEntry>  
26.     <systemUsage>      
27.         <systemUsage sendFailIfNoSpace="true">      
28.            <memoryUsage>    
29.                      
30.              <memoryUsage limit="6gb"/>   
31.            </memoryUsage>      
32.         </systemUsage>      
33.     </systemUsage>   
34. </broker>

**八. PendingSubscriberMessageStoragePolicy:(Topic)**

针对“非耐久”订阅者。概念和(七)一样，支持三种策略：storeCursor, vmCursor和fileCursor。

**九. PendingDurableSubscriberMessageStoragePolicy: (Topic)**

针对“耐久”订阅者，支持三种策略：storeDurableSubscriberCursor, vmDurableCursor和 fileDurableSubscriberCursor。

Java代码   ![收藏代码][icon_star.png]

1.  <policyEntry topic=">" producerFlowControl="false" memoryLimit="32mb">    
2.      <pendingSubscriberPolicy>    
3.              
4.           <fileCursor/>    
5.      </pendingSubscriberPolicy>    
6.      <pendingDurableSubscriberPolicy>    
7.              
8.              
9.              
10.             
11.         <storeDurableSubscriberCursor/>    
12.     </pendingDurableSubscriberPolicy>    
13. </policyEntry>

**十. ForcePersistencyModeBrokerPlugin:** 消息传输模式强制转换插件

将broker接收的消息强制转换成"PERSISTENT"或者"NOT\_PERSISTENT"。

Java代码   ![收藏代码][icon_star.png]

1.  <broker>  
2.      <plugins>  
3.            
4.        <forcePersistencyModeBrokerPlugin persistenceFlag="true"/>  
5.      </plugins>  
6.  </broker>

[icon_star.png]: /images/20220613/aa0e4fd671ee4e0e8d1193f7622a7581.png