activemq-集群和主从模式学习笔记

忘是亡心i 2022-06-17 05:19 311阅读 0赞

### activemq主从模式与集群 ###

#### 三种masterslave模式 ####

[三种主从模式][Link 1]

<table> 
 <thead> 
  <tr> 
   <th>主从方式</th> 
   <th>要求</th> 
   <th>性能</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>共享文件系统</td> 
   <td>要求SAN网络共享存储</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>JDBC主从方式</td> 
   <td>共享数据库</td> 
   <td>速度不如日志快</td> 
  </tr> 
  <tr> 
   <td>复制的LevelDB</td> 
   <td>zookeeper服务器</td> 
   <td>速度快</td> 
  </tr> 
 </tbody> 
</table>

**共享文件系统配置**  
多个broker争夺共享文件锁，取得文件锁的成为master

**JDBC主从方式**  
跟共享文件系统的方式类似，但是更可靠，速度上最慢

**可复制的LevelDB存储（zooKeeper主从模式推荐模式)**  
从一系列的broker中取出一个broker作为master，从master的broker同步更新的数据到所有的slave broker。对于N个的broker，同步复制（N/2 + 1）机器后，异步复制其他slave。  
该模式下下面的配置应该是一样的

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>注释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>replicas</td> 
   <td>3</td> 
   <td>(replicas/2 + 1)数量避免集群宕机</td> 
  </tr> 
  <tr> 
   <td>securityToken</td> 
   <td></td> 
   <td>做复制时候的安全令牌</td> 
  </tr> 
  <tr> 
   <td>zkAddress</td> 
   <td>127.0.0.1:2181</td> 
   <td>使用逗号分隔一系列zk</td> 
  </tr> 
  <tr> 
   <td>zkPassword</td> 
   <td></td> 
   <td>连接到zk时候的密码</td> 
  </tr> 
  <tr> 
   <td>zkPath</td> 
   <td>/default</td> 
   <td>主从切换时候选举信息</td> 
  </tr> 
  <tr> 
   <td>zkSessionTimeout</td> 
   <td>2s</td> 
   <td>检测一个zk节点是否宕机</td> 
  </tr> 
  <tr> 
   <td>sync</td> 
   <td>quorum_mem</td> 
   <td>使用逗号分隔的一系列值,local_mem,local_disk,remote_mem,remote_disk,quonum_mem,quonum_disk，如果结合多种同样的话，disk优先取，quorum_mem等价于local_mem,remote_mem，quorum_disk等价于local_disk，remote_disk</td> 
  </tr> 
 </tbody> 
</table>

每个节点都需要不同的

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>注释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>bind</td> 
   <td>tcp://0.0.0.0:61619</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>hostname</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>weight</td> 
   <td>1</td> 
   <td>权重</td> 
  </tr> 
 </tbody> 
</table>

标准的LevelDB存储属性

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>注释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>directory</td> 
   <td>LevelDB</td> 
   <td>用于存储数据文件，不存在会自动创建</td> 
  </tr> 
  <tr> 
   <td>readThreads</td> 
   <td>10</td> 
   <td>read线程数量</td> 
  </tr> 
  <tr> 
   <td>logSize</td> 
   <td>104857600(100MB)</td> 
   <td>单个数据文件最大值</td> 
  </tr> 
  <tr> 
   <td>verifyChecksums</td> 
   <td>false</td> 
   <td>校验数据文件</td> 
  </tr> 
  <tr> 
   <td>paranoidChecks</td> 
   <td>false</td> 
   <td>存储error尽早抛出异常</td> 
  </tr> 
  <tr> 
   <td>indexFactory</td> 
   <td>org.fusesource.leveldbjni.JniDBFactory, org.iq80.leveldb.impl.Iq80DBFactory</td> 
   <td>LevelDB索引</td> 
  </tr> 
  <tr> 
   <td>indexMaxOpenFiles</td> 
   <td>1000</td> 
   <td>索引能使用openfile数量</td> 
  </tr> 
  <tr> 
   <td>indexBlockRestartInterval</td> 
   <td>16</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>indexWriteBufferSize</td> 
   <td>6291456(16MB)</td> 
   <td>写入到磁盘前内存索引数据大小</td> 
  </tr> 
  <tr> 
   <td>indexBlockSize</td> 
   <td>4096</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>indexCacheSize</td> 
   <td>268435456(256MB)</td> 
   <td>用于缓存索引快的内存</td> 
  </tr> 
  <tr> 
   <td>indexCompression</td> 
   <td>snappy</td> 
   <td>用于索引块的压缩策略，snappy或者none</td> 
  </tr> 
  <tr> 
   <td>logCompression</td> 
   <td>none</td> 
   <td>用于日志记录的压缩策略,snappy或者none</td> 
  </tr> 
 </tbody> 
</table>

##### **使用KahaDB作为存储** #####

5.4版本以后activemq推荐使用kahaDB作为默认的存储机制  
[kahadb在activemq的使用][kahadb_activemq]

KahaDB 的属性

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>注释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>archiveCorruptedIndex</td> 
   <td>false</td> 
   <td>true为归档毁坏的索引(不删除)</td> 
  </tr> 
  <tr> 
   <td>archiveDataLogs</td> 
   <td>false</td> 
   <td>true表示把消息数据文件移动到归档目录(不删除)</td> 
  </tr> 
  <tr> 
   <td>checkForCorruptJournalFiles</td> 
   <td>false</td> 
   <td>true表示在启动的时候检查毁坏文件并尝试恢复</td> 
  </tr> 
  <tr> 
   <td>checkpointInterval</td> 
   <td>5000</td> 
   <td>单位毫秒</td> 
  </tr> 
  <tr> 
   <td>checksumJournalFiles</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>cleanupInterval</td> 
   <td>30000</td> 
   <td>单位毫秒</td> 
  </tr> 
  <tr> 
   <td>compactAcksAfterNoGC</td> 
   <td>10</td> 
   <td>设置过大会影响性能，</td> 
  </tr> 
  <tr> 
   <td>compactAcksIgnoresStoreGrowth</td> 
   <td>false</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>concurrentStoreAndDispatchQueues</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>concurrentStoreAndDispatchTopics</td> 
   <td>false</td> 
   <td>推荐设置为false</td> 
  </tr> 
  <tr> 
   <td>directory</td> 
   <td>activemq-data</td> 
   <td>数据和日志文件目录</td> 
  </tr> 
  <tr> 
   <td>directoryArchive</td> 
   <td>null</td> 
   <td>消费后消息归档位置</td> 
  </tr> 
  <tr> 
   <td>enableAckCompaction</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>enableIndexWriteAsync</td> 
   <td>false</td> 
   <td>true表示异步更新索引</td> 
  </tr> 
  <tr> 
   <td>enableJournalDiskSyncs</td> 
   <td>true</td> 
   <td>废，请看journalDiskSyncStrategy</td> 
  </tr> 
  <tr> 
   <td>ignoreMissingJournalfiles</td> 
   <td>false</td> 
   <td>true表示忽略遗失的日志文件</td> 
  </tr> 
  <tr> 
   <td>indexCacheSize</td> 
   <td>10000</td> 
   <td>在内存中缓存的索引页数量</td> 
  </tr> 
  <tr> 
   <td>indexDirectory</td> 
   <td></td> 
   <td>不设置使用directory目录</td> 
  </tr> 
  <tr> 
   <td>indexWriteBatchSize</td> 
   <td>1000</td> 
   <td>批量写入的索引数量</td> 
  </tr> 
  <tr> 
   <td>journalDiskSyncInterval</td> 
   <td>1000</td> 
   <td>单位毫秒，同步磁盘时间间隔</td> 
  </tr> 
  <tr> 
   <td>journalDiskSyncStrategy</td> 
   <td>always</td> 
   <td>always ：periodic ：never ：</td> 
  </tr> 
  <tr> 
   <td>journalMaxFileLength</td> 
   <td>32mb</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>maxAsyncJobs</td> 
   <td>10000</td> 
   <td>队列存储最大的异步消息，应该与MessageProducer的数量一样</td> 
  </tr> 
  <tr> 
   <td>preallocationScope</td> 
   <td>entire_journal_async</td> 
   <td>预分配文件</td> 
  </tr> 
  <tr> 
   <td>preallocationStrategy</td> 
   <td>sparse_file</td> 
   <td>如何预先分配文件，sparse_file，设置文件长度，os_kernel_copy，委托给操作系统，zeros不设置文件长度</td> 
  </tr> 
  <tr> 
   <td>storeOpenWireVersion</td> 
   <td>11</td> 
   <td>5.9.0默认为6</td> 
  </tr> 
 </tbody> 
</table>

#### networkConnector模式 ####

[网络模式][Link 2]  
用于提供高可用，容灾的分布式队列和话题，客户端能够快速从宕机的broker切换到另外正常运作的broker。

##### networkConnector的配置 #####

1.  硬编码所有的networkConnector的元素(该模式也支持其他的传输协议，比如http)

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://activemq.org/config/1.0">
    
      <broker brokerName="receiver" persistent="false" useJmx="false">  
        <networkConnectors>
          <networkConnector uri="static:(tcp://localhost:62001)"/>
        </networkConnectors>
    
        <persistenceAdapter>
          <memoryPersistenceAdapter/>
        </persistenceAdapter>
    
       <transportConnectors>
          <transportConnector uri="tcp://localhost:62002"/>
        </transportConnectors>
      </broker>
    
    </beans>

1.  使用自动发现检测broker(multicast or rendezvous)

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://activemq.org/config/1.0">
    
      <broker name="sender" persistent="false" useJmx="false">  
        <networkConnectors>
          <networkConnector uri="multicast://default"/>
        </networkConnectors>
    
        <persistenceAdapter>
          <memoryPersistenceAdapter/>
        </persistenceAdapter>
    
      <transportConnectors>
          <transportConnector uri="tcp://localhost:0" discoveryUri="multicast://default"/>
        </transportConnectors>
      </broker>
    
    </beans>

network Connector的配置形式如下

multicast://address:port?transportOptions

<table> 
 <thead> 
  <tr> 
   <th>选项名称</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>group</td> 
   <td>default</td> 
   <td>起一个唯一名字避免多播阻塞</td> 
  </tr> 
  <tr> 
   <td>minmumWireFormatVersion</td> 
   <td>0</td> 
   <td>最小允许的wireFomat版本</td> 
  </tr> 
  <tr> 
   <td>trace</td> 
   <td>false</td> 
   <td>记录传输的所有命令</td> 
  </tr> 
  <tr> 
   <td>useLocalHost</td> 
   <td>true</td> 
   <td>true表示本地机器命名为localhost</td> 
  </tr> 
  <tr> 
   <td>datagramSize</td> 
   <td>4 * 1024</td> 
   <td>UDP包的大小</td> 
  </tr> 
  <tr> 
   <td>timeToLive</td> 
   <td>-1</td> 
   <td>数据包生存时间，设置大于1可以发送到其他的broker中</td> 
  </tr> 
  <tr> 
   <td>loopBackMode</td> 
   <td>false</td> 
   <td>回路模式是否使用</td> 
  </tr> 
  <tr> 
   <td>wireFormat</td> 
   <td>default</td> 
   <td>使用的WireFormat名字</td> 
  </tr> 
  <tr> 
   <td>wireFormat.*</td> 
   <td></td> 
   <td>用于配置wireFormat</td> 
  </tr> 
 </tbody> 
</table>

使用默认值配置等价于multicast://239.255.2.3:6155

multicast://default

multicast://224.1.2.3:6255?group=mygroupnam

#### **OpenWire**\#\#\#\# ####

配置broker消息代理的信息  
需要加上前缀wireFormat.

<table> 
 <thead> 
  <tr> 
   <th>属性名</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>cacheEnabled</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>cacheSize</td> 
   <td>1024</td> 
   <td>当使用缓存时候，缓存的数量</td> 
  </tr> 
  <tr> 
   <td>maxInactivityDuration</td> 
   <td>30000</td> 
   <td>认为socket已经挂死的间隔时间然后主动杀死connection，为0表示不启用</td> 
  </tr> 
  <tr> 
   <td>maxInactivityDurationInitalDelay</td> 
   <td>10000</td> 
   <td>检查时延</td> 
  </tr> 
  <tr> 
   <td>maxFrameSize</td> 
   <td>MAX_LONG</td> 
   <td>允许的帧大小，配置以避免oom</td> 
  </tr> 
  <tr> 
   <td>prefixPacketSize</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>stackTraceEnabled</td> 
   <td>ture</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>tcpNoDelayEnabled</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>tightEncodingEnabled</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

##### **使用静态发现** #####

静态发现的属性

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>initialReconnectDelay</td> 
   <td>initialReconnectDelay</td> 
   <td>单位毫秒，useExponentialBackOff为false情况下，第一次尝试重连的时间间隔</td> 
  </tr> 
  <tr> 
   <td>maxReconnectDelay</td> 
   <td>30000</td> 
   <td>单位毫秒，重连间隔时间</td> 
  </tr> 
  <tr> 
   <td>useExponentialBackOff</td> 
   <td>true</td> 
   <td>每次重连失败都会增加重连时间间隔</td> 
  </tr> 
  <tr> 
   <td>backOffMultiplier</td> 
   <td>2</td> 
   <td>useExponentialBackOff每次增加时间</td> 
  </tr> 
 </tbody> 
</table>

uri="static:(tcp://host1:61616,tcp://host2:61616)?maxReconnectDelay=5000&useExponentialBackOff=false"

##### **MasterSlave discovery** #####

建立网络桥接模式到一个broker以及她的slave中，使用masterslave来进行操作

URI的顺序是  
The URIs are listed in order for: MASTER,SLAVE1,SLAVE2…SLAVE(thumbs down)

##### **NetworkConnector元素的属性** #####

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>name</td> 
   <td>bridge</td> 
   <td>network的名称，两个相同broker之间的不同networkConnector使用不同的名字</td> 
  </tr> 
  <tr> 
   <td>dynamicOnly</td> 
   <td>false</td> 
   <td>true表示仅当进行持久订阅的时候network才激活，默认在启动时候就激活</td> 
  </tr> 
  <tr> 
   <td>decreaseNetworkConsumerPriority</td> 
   <td>false</td> 
   <td>true表示消费者启动时候优先级为-5，经过一个跳的network连接消费者就减少1，false表示所有的消费者使用与本地消费者同样的优先级0</td> 
  </tr> 
  <tr> 
   <td>networkTTL</td> 
   <td>1</td> 
   <td>消息和持久订阅能够穿越的broker数量 both message&amp;consumer</td> 
  </tr> 
  <tr> 
   <td>messageTTL</td> 
   <td>1</td> 
   <td>消息能穿越的broker数量</td> 
  </tr> 
  <tr> 
   <td>consumerTTL</td> 
   <td>1</td> 
   <td>消费者能够穿越的broker数量</td> 
  </tr> 
  <tr> 
   <td>conduitSubscriptions</td> 
   <td>true</td> 
   <td>多个消费者消费消息被当作一个消费者</td> 
  </tr> 
  <tr> 
   <td>excludedDestinations</td> 
   <td>empty</td> 
   <td>在这个名单中的队列将不会在网络桥接模式在传递</td> 
  </tr> 
  <tr> 
   <td>dynamicallyIncludedDestinations</td> 
   <td>empty</td> 
   <td>可以在网络桥接模式的broker传播，empty表示不在excludedDestinations中的队列都可以传播</td> 
  </tr> 
  <tr> 
   <td>useVirtualDestSubs</td> 
   <td>false</td> 
   <td>true表示监听虚拟队列的公告消息</td> 
  </tr> 
  <tr> 
   <td>staticallyIncludedDestinations</td> 
   <td>empty</td> 
   <td>会在网络桥接中传播，即便没有消费者刚兴趣</td> 
  </tr> 
  <tr> 
   <td>duplex</td> 
   <td>false</td> 
   <td>是否全双工作模式，true同时用于生产和消费消息</td> 
  </tr> 
  <tr> 
   <td>prefetchSize</td> 
   <td>1000</td> 
   <td>每次预取的数量</td> 
  </tr> 
  <tr> 
   <td>suppressDuplicateQueueSubscriptions</td> 
   <td>false</td> 
   <td>true表示网络桥接中两个broker复制消息将会被制止，比如，A，B，C三个broker使用multicast discovery，A上的消费者a会在产生B和C的broker上，此外，C上消费者也会连接到B，B消费者也会连接到C；为true的时候，BC之间的网络桥接将被禁止，因为她们已经订阅了Abroker上面的消息了，减少路由选择，用这种方法能够确定消除消息在死路由（卡住消息）网络中迁移的可能，networkTTL大于等于broker的数量时候需要这个属性进行调停。</td> 
  </tr> 
  <tr> 
   <td>bridgeTempDestinations</td> 
   <td>true</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>alwaysSyncSend</td> 
   <td>false</td> 
   <td>true表示不持久化的消息使用请求应答的模式，这个配置对持久和非持久的选择都一样</td> 
  </tr> 
  <tr> 
   <td>staticBridge</td> 
   <td>false</td> 
   <td>true表示broker不会动态响应新的消费者，将会使用staticallyIncludedDestinations创建新的需求订阅。</td> 
  </tr> 
  <tr> 
   <td>userName</td> 
   <td>null</td> 
   <td>远程broker的用户名</td> 
  </tr> 
  <tr> 
   <td>password</td> 
   <td>null</td> 
   <td>密码</td> 
  </tr> 
 </tbody> 
</table>

**可靠性**  
如果源broker是持久订阅或者持久队列，network会保持持久化  
如果源broker是非持久化的，发生异常会导致线上消息丢失

**消息时序**  
无法保证消息时序，存在消费者从一个broker迁移到另外一个broker的情况。

**管道订阅（Conduit subscription）**  
activemq使用活跃的消费者在network中传递消息。broker复制了一份远程broker的订阅，然后路由传送给本地的客户端连接。如果是话题或者是超过一个的订阅，多个副本将会消息产生。导管就是为了预防副本消息在整个network中泛滥传播，这个默认运行情况，使得多个订阅在远程的broker上行为就如同单个订阅一样。  
然而，复制的订阅是一个有用的特性如果我们仅仅使用队列模式开发，这样子负载均衡算法就能把消息均匀的分配，如果conduitSubscriptions=false的话，在network上的消费者将会均分消息。比如AB两个broker，B上有2个消息者，A只有1个消费者，A生产了30个消息，在conduitSubscriptions=true的情况下，15个发送到A消费者，15个发送到B的borker上，如果conduitSubscriptions=false，10个发送到A，10个发送到B1，10个发送到B2。  
Duplex network connectors  
broker1配置到broker2的netwokrConnector是单向，如果设置duplex=true的话就是双向的了，  
同时可以配置在同一个broker中配置多个networkConnector增加吞吐量

##### **Conduit subscriptions and consumer selectors** #####

导管订阅忽略了在本地broker的消费者的selector并把消息传送给远程的消费者。远程的broker的selector在消息分发之前被解析，这在多broker中会存在问题。可以想象消息被A产生被传送到B或者C等broker，但是消息并不能满足B，C等broker等selector的特征，为了避免这种情况，要开启conduitSubscription这个属性。

##### **networkConnector的配置陷阱** #####

advisorySupport如果没有开启的话networkConnector可能会不正常的工作。  
networkConnector严重依赖于advisory messages，消费者连接到远程的broker的话，会接收到所有关于advisory Message，在大规模情况下工作效果不会工作的很好。  
只把特定队列的消息发送到远程broker。

在activemq的5.6版本或者更久远的需要增加前缀ActiveMQ.Advisory.Consumer.，配置demo如下

##### **Pure static networks** #####

用于保存broker不受远程broker的消费者的影响，或者把当前broker当做一个用于传送消息的代理，可以考虑配置成static network

staticBridge 参数在5.6版本以后使用，表示本地的broker不会从远程的broker订阅任何有关advisoryMessage，staticallyIncludedDestinations表示该队列或者是话题可以传播到远程的broker去，在5.6之前的版本不存在staticBridge的属性，我们可以destinationFilter 去监听没用到的advisory话题，类似下面的配置

如果像上述配置的话，broker将会尝试监听在ActiveMQ.Advisory.Consumer.NO\_DESTINATION的新的消费者，并且不受远程broker的消费者影响。

##### **Dynamic networks and Virtual Destinations**  #####

这里存在两个broker，LocalBroker配置的network如下

remote Broker配置了组合话题如下

假设在remote Broker的include.bar.forward队列的存在一个消费者。往include.bar话题发送了一个消息的话，实际上将会被传送到  
include.bar.forward队列并被消费掉。如果是往localBroker发布一个相同的话题，这个消息将不会传送到remoteBroker去。  
消息之所以不会被传送是因为在include.bar.forward的消费者不能检测到localBroker的dynamicallyIncludedDestinations 列表，因为没有在原始队列（话题）include.bar的消费者，消息不会被传送到RemoteBroker。这里我们通过配置LocalBroker来监听虚拟目的地订阅来解决。

首先，当消费者的订阅符合虚拟目的地，我们需要配置remoteBroker发送advisoryMessage，在这个例子中，我们配置remoteBroker的useVirtualDestSubs 为true

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://activemq.org/config/1.0">
    
      <broker name="remoteBroker" useVirtualDestSubs="true">  
         .....
      </broker>
    
    </beans>

同时，配置localBroker监听advisoryMessage设置useVirtualDestSubs 为true

配置后在remoteBroker的队列include.bar.forward的消费者可以获取到从LocalBroker传送过来的include.bar的话题。

##### **Virtual Destination Consumers on Destination Creation** #####

让我们重新考虑上面的组合情况，如果没有消费者在remoteBroker的队列上，  
remoteBroker的组合队列配置如下

组队话题配置在remoteBroker上，然后localBroker使用network连接到这个上面，即便我们开启了useVirtualDestSubs属性，消息也不会发送到没有消费者的remoteBroker上，这样子消息只会挤压在LocalBroker上，需要  
设置remoteBroker的属性useVirtualDestSubsOnCreation=true，消息会发送到remoteBroker上，同时remoteBroker必须是  
1到多个队列，拥有持久化的话题

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://activemq.org/config/1.0">
    
      <broker name="remoteBroker" useVirtualDestSubs="true" useVirtualDestSubsOnCreation="true">  
         .....
      </broker>
    
    </beans>

##### **Composite Destination consumers and Virtual Topics**  #####

LocalBroker

RemoteBroker

<?xml version="1.0" encoding="UTF-8"?>
    
    <beans xmlns="http://activemq.org/config/1.0">
    
      <broker name="remoteBroker" useVirtualDestSubs="true" >  
         .....
      </broker>
    
    </beans>

##### **卡住的消息(Stuck Messages)** #####

默认情况，消息是不允许回溯会他来自的broker，这样子就可以避免消息的传递成为一个环。偶然的情况下，存在需要消息重新回溯到原有队列。比如在这种情景，有两个broker，其中一个broker宕机后，消费者路由到另外一个broker中，然后宕机的broker被重启，但是没有消费者在宕机的broker上进行消费，所以宕机的broker重启后上挤压的消息没被消费掉，一个解决的方法是配置rebalanceClusterClients=true属性，另一种解决方法是允许消息回溯回他原来的broker。配置conditionalNetworkBridgeFilterFactory 标签的属性replayWhenNoConsumers=true。  
如果是activemq版本小于5.9，需要关闭游标复制检测，enableAudit=false。conditionalNetworkBridgeFilterFactory 还可以调节消费速率限制。

<destinationPolicy>
      <policyMap>
        <policyEntries>
          <policyEntry queue="TEST.>" enableAudit="false">
            <networkBridgeFilterFactory>
              <conditionalNetworkBridgeFilterFactory replayWhenNoConsumers="true"/>
            </networkBridgeFilterFactory>
          </policyEntry>
        </policyEntries>
      </policyMap>
    </destinationPolicy>

[Link 1]: http://activemq.apache.org/masterslave.html
[kahadb_activemq]: http://activemq.apache.org/kahadb.html
[Link 2]: http://activemq.apache.org/networks-of-brokers.html