ActiveMQ学习笔记目的地特性

拼搏现实的明天。 2022-06-16 08:41 274阅读 0赞

## 组合目的地 ##

使用单一的虚拟目的地往目的地集合发送消息，用一个虚拟的队列网多个物理队列发送消息，这里使用到,这个特殊的分割符。

// send to 3 queues as one logical operation
    Queue queue = new ActiveMQQueue("FOO.A,FOO.B,FOO.C");
    producer.send(queue, someMessage);

如果我们要混合不同类型的目的地，我们可以使用前缀queue://或者 topic:// 去区分不同的目的地类型

// send to queues and topic one logical operation
    Queue queue = new ActiveMQQueue("FOO.A,topic://NOTIFY.FOO.A");
    producer.send(queue, someMessage);

组合目的能够子broker上配置。

## 配置启动目的地 ##

## 删除不活动队列 ##

每schedulePeriodForDestinationPurge毫秒检查一次不活动的目的地，broker将会删除目的地该目的地已经超过inactiveTimoutBeforeGC毫秒是空的目的地且gcInactiveDestinations为 true的目的起。

<broker xmlns="http://activemq.apache.org/schema/core" schedulePeriodForDestinationPurge="10000">
    
      <destinationPolicy>
         <policyMap>
            <policyEntries>
               <policyEntry queue=">" gcInactiveDestinations="true" inactiveTimoutBeforeGC="30000"/>
            </policyEntries>
         </policyMap>
      </destinationPolicy>
    
    </broker>

## 目的地选项 ##

目的地选项是为JMS的消费者提供了可扩展的配置选项，可配置的参数是配置目的地的名字上的。

<table> 
 <thead> 
  <tr> 
   <th>选项</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>consumer.dispatchAsync</td> 
   <td>true</td> 
   <td>broker以异步的形式传递消息</td> 
  </tr> 
  <tr> 
   <td>consumer.exclusive</td> 
   <td>false</td> 
   <td>是否execlude消费者</td> 
  </tr> 
  <tr> 
   <td>consumer.maximumPendingMessageLimit</td> 
   <td>0</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer.noLocal</td> 
   <td>false</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer.prefetchSize</td> 
   <td>n/a</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer.priority</td> 
   <td>0</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer.retroactive</td> 
   <td>false</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer.selector</td> 
   <td>null</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

queue = new ActiveMQQueue("TEST.QUEUE?consumer.dispatchAsync=false&consumer.prefetchSize=10");
    consumer = session.createConsumer(queue);

## 镜像队列 ##

为了能够方便的监控队列，我们增加了叫做mirroredQueue的特性。  
假设现在我们存在队列Foo.Bar，已经开启了镜像队列的功能，那么我们可以订阅话题VirtualTopic.Mirror.Foo.Bar，这样子我们就可以接受到所有发送到队列Foo.bar的消息，因为镜像队列就是复制了一份原来队列的消息发送到镜像队列去。我嗯也可以使用Virtual Topic的特性，订阅队列Consumer.A.VirtualTopic.Mirror.Foo.Bar ，那么所有的消息将会传输到消费者A去。  
镜像队列默认情况下是关闭的，可以在BrokerService里面开启useMirroredQueues的属性，或者在broker内部中配置，下面的配置还配置了通配

## 目的地属性策略 ##

<table> 
 <thead> 
  <tr> 
   <th>公共属性</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>advisoryForConsumed</td> 
   <td>false</td> 
   <td>当消息被消费的时候发送公告信息</td> 
  </tr> 
  <tr> 
   <td>advisoryForDelivery</td> 
   <td>false</td> 
   <td>当消息被发送到client时候发送一个公告信息</td> 
  </tr> 
  <tr> 
   <td>advisoryForFastProducers</td> 
   <td>false</td> 
   <td>当生产者被认为是快的发送一个公告消息</td> 
  </tr> 
  <tr> 
   <td>advisoryForSlowConsumers</td> 
   <td>false</td> 
   <td>当消费者被认为是慢的时候发送公告消息</td> 
  </tr> 
  <tr> 
   <td>advisoryWhenFull</td> 
   <td>false</td> 
   <td>当超过极限（内存，存储，swap）发送一个公告</td> 
  </tr> 
  <tr> 
   <td>cursorMemoryHighWaterMark</td> 
   <td>70</td> 
   <td>systemUsage与memoryUsage的百分比，或者是目的地的内存限制，超过该阈值将会导致游标写阻塞或者写磁盘</td> 
  </tr> 
  <tr> 
   <td>enableAudit</td> 
   <td>true</td> 
   <td>true表示borker将会追踪消息副本，在快速恢复中非持久的消息将会发生复制</td> 
  </tr> 
  <tr> 
   <td>gcInactiveDestinations</td> 
   <td>false</td> 
   <td>回收不活动的目的地</td> 
  </tr> 
  <tr> 
   <td>gcInactiveDestinations</td> 
   <td>5000</td> 
   <td>单位毫秒，多久才认为一个队列是不活动的</td> 
  </tr> 
  <tr> 
   <td>includeBodyForAdvisory</td> 
   <td>false</td> 
   <td>用于保存消息的dataStructure</td> 
  </tr> 
  <tr> 
   <td>maxBrowsePageSize</td> 
   <td>400</td> 
   <td>浏览器页面打开的最大数量</td> 
  </tr> 
  <tr> 
   <td>maxDestinations</td> 
   <td>-1</td> 
   <td>大于0用于限制创建队列数量</td> 
  </tr> 
  <tr> 
   <td>maxPageSize</td> 
   <td>200</td> 
   <td>一次从存储中获取消息的最大数量，增加这个数量用于提升性能</td> 
  </tr> 
  <tr> 
   <td>memoryLimit</td> 
   <td>n/a</td> 
   <td>目的地游标的内存限制，该限制低级于系统级别的内存限制，即systemUsage ／ memoryUsage属性，这里不存在默认值，内存限制就是borker的全部内存，该属性使用时，cursorMemoryHighWaterMark也将会启用，systemUsage／memoryUsage属性将会被禁用</td> 
  </tr> 
  <tr> 
   <td>minimumMessageSize</td> 
   <td>1024</td> 
   <td>无法序列化消息的消息大小，可序列化的消息的消息可以在序列化制定，无需使用该值</td> 
  </tr> 
  <tr> 
   <td>prioritizedMessages</td> 
   <td>false</td> 
   <td>持久话优先级高的消息</td> 
  </tr> 
  <tr> 
   <td>producerFlowControl</td> 
   <td>true</td> 
   <td>表示对消息的产生数量进行控制（通过不响应生产者的ACK或者是抛出javax.jms.ResourceAllocationException）当内存或者存储已经耗尽，false表示额外的消息将会被写入到磁盘中避免内存耗尽，在存储达到最大容量的时候也会对其进行流量控制。</td> 
  </tr> 
  <tr> 
   <td>storeUsageHighWaterMark</td> 
   <td>100</td> 
   <td>systemUsage与storeUsage的百分比限制，超过会导致发送阻塞</td> 
  </tr> 
  <tr> 
   <td>useCache</td> 
   <td>true</td> 
   <td>持久话消息就会被缓存</td> 
  </tr> 
  <tr> 
   <td>usePrefetchExtension</td> 
   <td>true</td> 
   <td>当消息被传递但不被ACK时使用预取扩展，使得代理可以调度另一个消息，例如预取== 0，这个想法是总是存在待预取的消息的预取数。 它还允许事务批处理超过预取值。</td> 
  </tr> 
 </tbody> 
</table>

队列的额外属性

<table> 
 <thead> 
  <tr> 
   <th>队列属性</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>allConsumersExclusiveByDefault</td> 
   <td>false</td> 
   <td>所有的消费者就会是exclusive的</td> 
  </tr> 
  <tr> 
   <td>consumersBeforeDispatchStarts</td> 
   <td>0</td> 
   <td>第一个消费者连上时候，等到制定数量消费者连上后才开始分发消息</td> 
  </tr> 
  <tr> 
   <td>expireMessagesPeriod</td> 
   <td>30000</td> 
   <td>单位毫秒，周期的检查消息是否过期，0表示不检查</td> 
  </tr> 
  <tr> 
   <td>lazyDispatch</td> 
   <td>false</td> 
   <td>从存储中读取的消息才能分发</td> 
  </tr> 
  <tr> 
   <td>optimizedDispatch</td> 
   <td>false</td> 
   <td>不使用单独的线程分发消息</td> 
  </tr> 
  <tr> 
   <td>persistJMSRedelivered</td> 
   <td>false</td> 
   <td>true表示</td> 
  </tr> 
 </tbody> 
</table>

**队列专有属性**

<table> 
 <thead> 
  <tr> 
   <th>队列专有属性</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>allConsumersExclusiveByDefault</td> 
   <td>false</td> 
   <td>所有的消费者将会是exclusive的</td> 
  </tr> 
  <tr> 
   <td>consumersBeforeDispatchStarts</td> 
   <td>0</td> 
   <td>等待足够的消费者才分发消息</td> 
  </tr> 
  <tr> 
   <td>expireMessagesPeriod</td> 
   <td>30000</td> 
   <td>单位毫秒，周期检查消息是否过期，0表示不检查</td> 
  </tr> 
  <tr> 
   <td>lazyDispatch</td> 
   <td>false</td> 
   <td>从持久设备加载到内存页面的消息立即分发</td> 
  </tr> 
  <tr> 
   <td>optimizedDispatch</td> 
   <td>false</td> 
   <td>不使用多线程分发消息</td> 
  </tr> 
  <tr> 
   <td>persistJMSRedelivered</td> 
   <td>false</td> 
   <td>true表示持久话消息被broker第一次分发前，消息会被重写反应出可能的传送</td> 
  </tr> 
  <tr> 
   <td>queuePrefetch</td> 
   <td>n/a</td> 
   <td>设置预取的默认值</td> 
  </tr> 
 </tbody> 
</table>

## 虚拟目的地 ##

### 持久化JMS话题的限制 ###

JMS创建的持久化消费者拥有唯一的JMS的clientID和名字，一个JMS的connection同时有且只能拥有一个JMSclientID，而一个消费者仅能被一个clientId激活，也就是说，仅仅只有一个线程能勾葱给定的逻辑话题中消费消息，故我们不能实现

*  负载均衡
 *  当某个线程挂死的时候我们不能进行快速恢复

**虚拟话题的配置**

<destinationInterceptors>
      <virtualDestinationInterceptor>
        <virtualDestinations>
          <virtualTopic name=">" prefix="VirtualTopicConsumers.*." selectorAware="false"/>
        </virtualDestinations>
      </virtualDestinationInterceptor>
    </destinationInterceptors>

<table> 
 <thead> 
  <tr> 
   <th>选项</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>selectorAware</td> 
   <td>false</td> 
   <td>当消息匹配到存在的订阅者就分发</td> 
  </tr> 
  <tr> 
   <td>local</td> 
   <td>false</td> 
   <td>不扇出葱network接收到的消息</td> 
  </tr> 
  <tr> 
   <td>concurrentSend</td> 
   <td>false</td> 
   <td>true表示使用线程池发送消息，允许日志批量写日志以减少磁盘io</td> 
  </tr> 
  <tr> 
   <td>transactedSend</td> 
   <td>false</td> 
   <td>true表示单个线程进程磁盘同步，如果不存在事务将会开启一个事务</td> 
  </tr> 
 </tbody> 
</table>

**组合目的地**

下面配置的MY.QUEUE的队列将会把自己接受到消息转发到FOO队列和BAR话题，需要知道的事MY.QUEUE只是一个逻辑上的队列，订阅者消费消息仅仅只能从FOO和BAR消费消息，而不是MY.QUEUE。

但是逻辑的队列的这个行为是可以改变的，将forwardOnly设置为false，例如，IncomingOrders队列的消息在被IncomingOrders的订阅者消费之前将会被复制并被发送到Notifications队列。  
当forwardOnly属性不设置或者为true的时候，compositeQueue和compositeTopic没有逻辑上的不同，它们可以等价交换使用。

### 使用过滤目的 ###

ActiveMQ4.2版本我们可以使用选择器去定义虚拟目的地。

### 避免network broker模式下的消息重复 ###

如果我们在network broker中使用虚拟话题，使用默认的network配置将会得到重复的消息，因为网络节点不仅会把消息发送到虚拟话题，而且也会关联到物理队列，所以在配置的时候我们要禁止消息在物理队列上的转发。

## 通配符 ##

通配符不是JMS规范的一部分

*  .表示路径分隔符
 *  \*表示匹配任意名称在路径中
 *  > 表示匹配任意名称  
    > 需要注意的是，不要在>后面跟随字符串，因为后面的字符串将会被忽略。
    
    例子

<table> 
 <thead> 
  <tr> 
   <th>订阅</th> 
   <th>意义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>PRICE.&gt;</td> 
   <td>在所有交易系统的产品价格</td> 
  </tr> 
  <tr> 
   <td>PRICE.STOCK.&gt;</td> 
   <td>所有交易系统的股票价格</td> 
  </tr> 
  <tr> 
   <td>PRICE.STOCK.NASDAQ.*</td> 
   <td>在NASDAQ交易系统中的股票价格</td> 
  </tr> 
  <tr> 
   <td>PRICE.STOCK.*.IBM</td> 
   <td>IBM在所有交易系统中的股票价格</td> 
  </tr> 
 </tbody> 
</table>