学步园

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，并采用如下的配置文件：

Xml代码

1. <
   beans
   >
   
   
     
   
2.   
   <
   broker
   
   
    
   xmlns
   
   =
   "http://activemq.org/config/1.0"
   
    
   brokerName
   
   =
   "BugBroker1"
   
    
   useJmx
   
   =
   "true"
   
   >
   
   
     
3.      
4.     
   <
   transportConnectors
   >
   
   
     
5.       
   <
   transportConnector
   
   
    
   uri
   
   =
   "tcp://localhost:61616"
   
   />
   
   
     
6.     
   </
   transportConnectors
   >
   
   
     
7.        
8.     
   <
   persistenceAdapter
   >
   
   
     
9.       
   <
   amqPersistenceAdapter
   
   
    
   directory
   
   =
   "activemq-data/BugBroker1"
   
    
   maxFileLength
   
   =
   "32mb"
   
   />
   
   
     
10.     
    </
    persistenceAdapter
    >
    
    
      
11.         
12.   
    </
    broker
    >
    
    
      
13. </
    beans
    >
    
    
      

<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，如下：

Xml代码

1. <
   broker
   
   
    
   brokerName
   
   =
   "receiver"
   
    
   persistent
   
   =
   "false"
   
    
   useJmx
   
   =
   "false"
   
   >
   
   
     
   
2.   
   <
   transportConnectors
   >
   
   
     
3.     
   <
   transportConnector
   
   
    
   uri
   
   =
   "tcp://localhost:62002"
   
   />
   
   
     
4.   
   </
   transportConnectors
   >
   
   
     
5.   
   <
   networkConnectors
   >
   
   
     
6.     
   <
   networkConnector
   
   
    
   uri
   
   =
   "static:( tcp://localhost:61616,tcp://remotehost:61616)"
   
   />
   
   
     
7.   
   </
   networkConnectors
   >
   
   
     
8.   …   
9. </
   broker
   >
   
   
     

<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，如下：

Xml代码

1. <
   broker
   
   
    
   name
   
   =
   "sender"
   
    
   persistent
   
   =
   "false"
   
    
   useJmx
   
   =
   "false"
   
   >
   
   
     
   
2.   
   <
   transportConnectors
   >
   
   
     
3.     
   <
   transportConnector
   
   
    
   uri
   
   =
   "tcp://localhost:0"
   
    
   discoveryUri
   
   =
   "multicast://default"
   
   />
   
   
     
4.   
   </
   transportConnectors
   >
   
   
     
5.   
   <
   networkConnectors
   >
   
   
     
6.     
   <
   networkConnector
   
   
    
   uri
   
   =
   "multicast://default"
   
   />
   
   
     
7.   
   </
   networkConnectors
   >
   
   
     
8.   ...   
9. </
   broker
   >
   
   
     

<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有以下属性：

   
关于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属
性的一个例子：

Xml代码

1. <
   networkConnectors
   >
   
   
     
   
2.   
   <
   networkConnector
   
   
    
   uri
   
   =
   "static://(tcp://localhost:61617)"
   
     
3.      
   name
   
   =
   "bridge"
   
    
   dynamicOnly
   
   =
   "false"
   
    
   conduitSubscriptions
   
   =
   "true"
   
     
4.      
   decreaseNetworkConsumerPriority
   
   =
   "false"
   
   >
   
   
     
5.      
   <
   excludedDestinations
   >
   
   
     
6.        
   <
   queue
   
   
    
   physicalName
   
   =
   "exclude.test.foo"
   
   />
   
   
     
7.        
   <
   topic
   
   
    
   physicalName
   
   =
   "exclude.test.bar"
   
   />
   
   
     
8.      
   </
   excludedDestinations
   >
   
   
     
9.      
   <
   dynamicallyIncludedDestinations
   >
   
   
     
10.        
    <
    queue
    
    
     
    physicalName
    
    =
    "include.test.foo"
    
    />
    
    
      
11.        
    <
    topic
    
    
     
    physicalName
    
    =
    "include.test.bar"
    
    />
    
    
      
12.      
    </
    dynamicallyIncludedDestinations
    >
    
    
      
13.      
    <
    staticallyIncludedDestinations
    >
    
    
      
14.        
    <
    queue
    
    
     
    physicalName
    
    =
    "always.include.queue"
    
    />
    
    
      
15.        
    <
    topic
    
    
     
    physicalName
    
    =
    "always.include.topic"
    
    />
    
    
      
16.      
    </
    staticallyIncludedDestinations
    >
    
    
      
17.   
    </
    networkConnector
    >
    
    
      
18. </
    networkConnectors
    >
    
    
      

<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推荐的高可靠性和容错的解决方案。以下是几种不同的类型：

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需要进行以下配置

Xml代码

1. <
   broker
   
   
    
   masterConnectorURI
   
   =
   "tcp://masterhost:62001"
   
    
   shutdownOnMasterFailure
   
   =
   "false"
   
   >
   
   
     
   
2.     ...   
3.     
   <
   transportConnectors
   >
   
   
     
4.       
   <
   transportConnector
   
   
    
   uri
   
   =
   "tcp://slavehost:61616"
   
   />
   
   
     
5.    
   </
   transportConnectors
   >
   
   
     
6. </
   broker
   >
   
   
     

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

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

Xml代码

1. <
   broker
   
   
    
   brokerName
   
   =
   "slave"
   
    
   useJmx
   
   =
   "false"
   
     
   deleteAllMessagesOnStartup
   
   =
   "true"
   
     
   xmlns
   
   =
   "http://activemq.org/config/1.0"
   
   >
   
   
     
   
2.   ...   
3.   
   <
   services
   >
   
   
     
4.     
   <
   masterConnector
   
   
    
   remoteURI
   
   = 
   "tcp://localhost:62001"
   
    
   userName
   
   =
   "user"
   
    
   password
   
   =
   "password"
   
   />
   
   
     
5.   
   </
   services
   >
   
   
     
6. </
   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配置的一个例子：

Xml代码

1. <
   broker
   
   
    
   useJmx
   
   =
   "false"
   
     
   xmlns
   
   =
   "http://activemq.org/config/1.0"
   
   >
   
   
     
   
2.    
   <
   persistenceAdapter
   >
   
   
     
3.       
   <
   journaledJDBC
   
   
    
   dataDirectory
   
   =
   "/sharedFileSystem/broker"
   
   />
   
   
     
4.    
   </
   persistenceAdapter
   >
   
   
     
5.    …   
6. </
   broker
   >
   
   
     

<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配置的一个例子：

Xml代码

1. <
   beans
   >
   
   
     
   
2.   
   <
   broker
   
   
    
   xmlns
   
   =
   "http://activemq.org/config/1.0"
   
    
   brokerName
   
   =
   "JdbcMasterBroker"
   
   >
   
   
     
3.     ...   
4.     
   <
   persistenceAdapter
   >
   
   
     
5.       
   <
   jdbcPersistenceAdapter
   
   
    
   dataSource
   
   =
   "#mysql-ds"
   
   />
   
   
     
6.     
   </
   persistenceAdapter
   >
   
   
     
7.        
8.   
   </
   broker
   >
   
   
     
9.      
10.   
    <
    bean
    
    
     
    id
    
    =
    "mysql-ds"
    
     
    class
    
    =
    "org.apache.commons.dbcp.BasicDataSource"
    
     
    destroy-method
    
    =
    "close"
    
    >
    
    
      
11.     
    <
    property
    
    
     
    name
    
    =
    "driverClassName"
    
     
    value
    
    =
    "com.mysql.jdbc.Driver"
    
    />
    
    
      
12.     
    <
    property
    
    
     
    name
    
    =
    "url"
    
     
    value
    
    =
    "jdbc:mysql://localhost:3306/test?relaxAutoCommit=true"
    
    />
    
    
      
13.     
    <
    property
    
    
     
    name
    
    =
    "username"
    
     
    value
    
    =
    "username"
    
    />
    
    
      
14.     
    <
    property
    
    
     
    name
    
    =
    "password"
    
     
    value
    
    =
    "passward"
    
    />
    
    
      
15.     
    <
    property
    
    
     
    name
    
    =
    "poolPreparedStatements"
    
     
    value
    
    =
    "true"
    
    />
    
    
      
16.   
    </
    bean
    >
    
    
        
17. </
    beans
    >
    
    
      

<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代码

1. ALTER
   
   
    
   TABLE
   
   
    activemq_acks ENGINE = InnoDB;
     
   
2. ALTER
   
   
    
   TABLE
   
   
    activemq_lock ENGINE = InnoDB;
     
3. ALTER
   
   
    
   TABLE
   
   
    activemq_msgs ENGINE = InnoDB;