第九章 ActiveMq的高级特性

                      第九章 ActiveMq的高级特性

 

章节导读

•    配置ActiveMQ的高可用
•    理解网络代理(networks of brokers)
•    通过配置扩展ActiveMq

      

 

 9.1 配置ActiveMq高可用

        

      当应用被部署到生产环境中时,你需要考虑到容灾问题(网络,硬件,软件).ActiveMq可以防御性的部署让你在那些问题发生时保证可用.一般来说,你需要在不同的机子上运行多个Activemq brokers,以防其中某一台机子或者某一个broker失效的时候其它的可以接替.这种叫做master/slave(主从)架构,其中某一个broker作为主节点,其它的从节点等待主节点失效时,其中的某一个从节点会作为主节点的替代.Actimve Java和C++客户端都提供了失效重连机制(failover),在消息不丢失的情况下让他们可以自动的从失效的主节点切换到新的节点.

      ActiveMq当前支持两种形式的主从配置:

•     不共享数据 : 每个broker有它自己的消息存储
•   共享数据 : 多个broker通过连接来共享消息存储(关联数据库或者文件共享系统)

    9.1.1 不共享数据的主从配置

 

         这应该是最简单用于提供高可用的消息代理的方式,一个从节点被被配置连接到主节点.从节点需要一个特殊的配置来表示它的特殊状态,所有消息命令都会被复制到从节点,复制发生在主节点接受命令之前.从节点会在启动的时候连接到主节点,所有主节点应该先启动,从节点不会开启任何传输协议,本身不会开启任何网络连接除非主节点挂了.从节点失去了和主节点的连接之后就会认为主节点挂了.

        这种模式的优点就是开销小,当一个生产者发送一条持久化信息给主节点,直到主节点接收到消息之后它才能发送下一条消息.

        直到消息被复制到从节点,并且从节点完成对消息的处理(持久化之类的),主节点才会处理这条消息(持久化并发送).接着就会发送一个响应告诉生产者消息已经被成功处理了.

       当主节点失效,从节点有两个选择

• 自己关闭
• 开启传输协议并初始化网络连接

       不共享数据的这种方式有一些限制.主节点只会在从节点连接到它的那个时候复制.如果一个客户端在从节点连接到主节点之前连接,那么主节点挂了消息就会丢失.为了避免这种情况,可以在主节点中设置wairForSlave属性.这个属性强制主节点在从节点连接到它之前,不接受任何连接.另外一个限制就是只允许有一个从节点，并且从节点不能有另外的从节点.

      如果你正在运行一个broker,你想要给它加入一个从节点,你需要停止它的服务,拷贝所有消息存储文件给从节点.主节点挂了之后从节点取而代之,之后你也需要把从节点的数据拷贝到主节点.这种模式能确保消息不会丢失,但是你必须接受服务需要停止一段时间用来拷贝数据.

                      

         配置从节点

               

<services>
       <masterConnector remoteURI="tcp://remotehost:62001"  userName="Rob" password="Davies"/>
</services>

      remoteURI：主节点监听的端口地址

 

 userName，password:如果住节点有配置认证,需要设置

  shutdownOnMasterFailure:false(默认)

     主节点属性:

       waitForSlave:false       等待从节点连接
       shutdownOnSlaveFailure:false   从节点挂了跟着挂

 

   9.1.1 不共享数据的主从配置

 

         这种模式允许许多brokers共享storage,但是在任何时间只能有一个broker访问它.使用这种模式可以确保主节点挂了之后,不需要做额外的操作.也没有从节点数量的限制.

        可以用两种方式来实现

        数据库

        它通过数据库的独占锁来实现在同一时间只有一个broker能访问数据库表.当主节点挂了之后,某一个从节点会获取数据库锁.

        这种模式会比不共享模式总体来说要慢,但是它不需要额外的配置,对从节点的数量也没有限制.如果你对性能有很高的要求,可以用文件共享系统代替.

        

 

     文件共享系统

 

      使用kahaDb消息存储,当kahaDb消息存储启动时,它会尝试获取文件锁,防止其它进程访问.第一个获取锁的broker自动成为master.

      文件共享系统需要一个语义上的分布式共享文件锁,如果你没有在使用SAN(storage area network),NFS(storage area network)可以作为替代.如果你使用Fedora或者RedHat(5.3或更高),你需要使用GFS(global file system) 2(需要集群锁协议,例如dlm,分布式锁管理器,linux的内核模块),

      使用文件共享系统是ActimveMq高可用的最好实现方式.它结合了KahaBd的高吞吐量和获取共享资源的便利性,KahaDb的性能取决于底层的文件共享系统

   

 

  9.2 ActiveMq如何通过网络代理传递消息

 

          activeMq支持把broker连接到不同的拓扑结构,或者已知的网络代理.这个需要一种分散的应用程序可以可靠的通信.

          

          9.2.1 存储和转发

 

           activeMq网络使用存储和转发的概念,.在通过网络被转发到其他代理之前消息总是被存储在本地代理中.这意味着连接出现问题的时候消息是不会被传递的,当连接恢复时,代理就会还没被发送的消息发送给远程代理.

      

 

          当远程代理和本地代理建立连接之后,远程代理会发送包含所有持久的和活跃的消费者的目的地的信息给本地代理.本地代理根据这些信息来决定远程的代理对哪些信息感兴趣并转发给他们.可以在网络连接中定义过滤器（包含或者排除某些目的地）. 默认网络操作都是单向的,如果你想转换成双向的,你可以给远程代理配置一个网络连接器指向本地代理,或者配置一个双攻的网络连接器.

 

           想象一个场景,有许多超市需要连接到后勤订单系统.配置新的超市和后勤系统的代理能识别每个新的超市对应的远程代理是很难的,后勤系统代理有可能在防火墙之后并且只开放限制数量的端口.

    

       
    上图中的网络连接是双工模式.这条网络连接可以双向的传送信息.

 

      超市代理的配置需要包含下面这样的网络连接器

        

<networkConnectors>
 <networkConnector uri="static://(tcp://backoffice:61617)"    name="bridge"  duplex="true"  conduitSubscriptions="true"  decreaseNetworkConsumerPriority="false">
   </networkConnector>
</networkConnectors>

 

 

       请注意指定网络连接和持久化配置的顺序是狠重要的.需要按下面的顺序配置

•    网络  ---在消息存储配置之前
•  消息存储  --- 在传输器之前配置
• 传输器 --- 最后配置.

   正确的配置如下边所示

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://activemq.apache.org/schema/core">
<broker brokerName="receiver" persistent="true" useJmx="true">
<networkConnectors>
<networkConnector uri="static:(tcp://backoffice:61617)"/>
</networkConnectors>
<persistenceAdapter>
<kahaDB directory = "activemq-data"/>
</persistenceAdapter>
<transportConnectors>
<transportConnector uri="tcp://localhost:62002"/>
</transportConnectors>
</broker>
</beans>

   

 这种方案,本地主节点和从节点都要配置远程主节点和从节点的网络连接,为了防止远程节点失效.

 

    9.2.2 发现网络

 

        当建立一个网络连接到一个远程代理节点,这个连接使用发现代理(discovery agent)来定位远程代理以及建立连接.activemq提供了两种形式发现网络的机制

•  动态的--使用广播和汇合机制来收搜代理
• 静态的--配置代理url列表.

    使用广播的方式来创建网络连接是最直接的,当你在网络连接器上开启广播,它就会用ip广播的方式建立连接,例如:

    

<networkConnectors>
   <networkConnector uri="multicast://default"/>
</networkConnectors>

  但是你需要定义一个唯一的组名,可以让它不会连接到其它不相关的应用代理.使用广播方式会有很多限制,例如不能控制哪个代理被发现,也只能发现本地网络段的代理,因为它不会通过路由.

 

  

  使用静态方式的例子如下:

 

<networkConnectors>
    <networkConnector  uri="static:(tcp://remote-master:61617,tcp://remote-slave:61617)"/>
</networkConnectors>

  它有如下的属性:

 

属性名	默认值	描述
initialReconnectDelay	1000	尝试重连网络之前的时间,只有当useExponentialBackOff不启用的时候它才起作用.
maxReconnectDelay	30000	在连接失败后,在试着重连之前的最大等待时间,useExponentialBackOff启用它才有效
useExponentialBackOff	true	如果启用,每次重连失败到再次重连之间的等待时间都会递增
backOffMultiplier	2	每次重连失败到再次重连之间的等待时间的倍数.与useExponentialBackOff配合使用.

   

  可以如下设置属性

   

<networkConnectors>
  <networkConnector  uri="static:(tcp://remote:61617)?useExponentialBackOff=false"/>
</networkConnectors>

 

 

  9.2.3 网络配置

 

     为了能在动态网络环境中正常运行,advisorySupport属性需要被启用,activemq使用咨询(advisory)信息在代理之间通信状态.咨询(advisory)信息用于通过代理网络和客户端转发关于改变消息的消费者的信息.也许在远程代理商没有持久的订阅者和消费者,所以当远程代理初始化网络连接时,远程代理将阅读消息存储为现有的目的地 并将这些传递给本地代理.然后本地代理就会为这些目的地转发消息.

    网络将会使用代理的名字来创建唯一的持久订阅代理来代表一个远程代理,如果之后你改变了代理的名字,你就会丢失订阅者,为了避免这个,确地使用唯一的brokername属性.例子如下:

   

<broker xmlns="http://activemq.apache.org/schema/core"
brokerName="brokerA"  dataDirectory="${activemq.base}/data">
...
  <networkConnectors>
    <networkConnector   name="brokerA-to-brokerB" uri="tcp://remotehost:61616" />
   </networkConnectors>
</broker>

    networkConnector节点有以下的属性

 

•     dynamicOnly  -- 如果为true, 持久订阅被激活时才创建对应的网路持久订阅。默认是启动时激活.dynamicOnly默认是false.
•    PREFETCHSIZE -- 预取多少条消息,消息确认总是使用INDIVIDUAL_ACK模式,默认值是1000.
• conduitSubscriptions  -- 多个消费者是否被当做一个消费者  默认true
• EXCLUDEDDESTINATIONS  -- 不通过网络转发的destination 默认empty
•     

  <networkConnectors>
  <networkConnector
  uri="static:(tcp://remote:61617)?useExponentialBackOff=false">
  <excludedDestinations>
    <queue physicalName="audit.queue-1"/>
    <queue physicalName="audit.queue-2"/>
    <queue physicalName="local.>"/>
    <topic physicalName="local.>"/>
  </excludedDestinations>
  </networkConnector>
  </networkConnectors>

   
• DYNAMICALLYINCLUDEDDESTINATIONS   --  通过网络转发的destinations，注意空列表代表所有的都转发.默认空,都转发
• STATICALLYINCLUDEDDESTINATIONS -- 匹配的都将通过网络转发-即使没有对应的消费者  默认 empty
• DECREASENETWORKCONSUMERPRIORITY -- 如果为true，网络的消费者优先级降低为-5,如果false,和本地消费者一样为0.默认false
• NETWORKTTL -- 消息和订阅在网络上通过的最大broker数量  默认1
• Name -- 名称.默认bridge.当在本地有两个和远程代理节点的连接时,这个是很有用的.
•  

  <networkConnectors>
  	<networkConnector uri="static://(tcp://remotehost:61617)" name="queues_only" duplex="true"
  			<excludedDestinations>
  				<topic physicalName=">"/>
  				</excludedDestinations>
  			</networkConnector>
  	<networkConnector uri="static://(tcp://remotehost:61617)" name="topics_only" duplex="true"
  			<excludedDestinations>
  				<queue physicalName=">"/>
  				</excludedDestinations>
  	</networkConnector>
  </networkConnectors>

  9.3 ActiveMq 处理大并发应用程序

 

        9.3.1 垂直扩展

              

              垂直扩展技术用于增加一个Activemq代理可以处理的连接数量.默认情况下,Activemq会使用阻塞IO处理传输连接,你也可以使用非阻塞IO的形式来减少线程的数量,非阻塞IO可以通过配置transport connector属性

             

<broker>
<transportConnectors>
   <transportConnector name="nio" uri="nio://localhost:61616"/>
</<transportConnectors>
</broker>

           还可以通过设置系统属性ACTIVEMQ_OPTS="-Dorg.apache.activemq.UseDedicatedTaskRunner=false"使用线程池.

 

          为了确保Activemq代理有足够的内存处理高并发程序.第一,你需要确保启动时有足够的内存,例如ACTIVEMQ_OPTS="-Xmx1024M \   -Dorg.apache.activemq.UseDedicatedTaskRunner=false".

          第二,要为activemq代理的jVm配置适当可用的内存大小,通过systemUsage节点的limit属性配置,一个好的做法就是,如果activemq有几百个连接时在开始时设置512M内存.如果不够再加.

        

<systemUsage>
	<systemUsage>
		<memoryUsage>
			<memoryUsage limit="512 mb"/>
		</memoryUsage>
		<storeUsage>
			<storeUsage limit="10 gb" name="foo"/>
		</storeUsage>
		<tempUsage>
			<tempUsage limit="1 gb"/>
		</tempUsage>
	</systemUsage>
</systemUsage>

         减轻每个连接的CPU负载也是很有必要的,如果使用openwire协议,禁用tight encoding属性,failover://(tcp://localhost:61616?wireFormat.tightEncodingEnabled=false)";

         默认队列配置使用一个单独的线程来分页消息存储到队列的消息,然后再发送给感兴趣的消息消费者.如果有大量的队列,建议开启optimizeDispatch属性来禁用这个特性.

       

<destinationPolicy>
	<policyMap>
		<policyEntries>
			<policyEntry queue=">" optimizedDispatch="true"/>
			</policyEntries>
		</policyMap>
	</destinationPolicy>

      为了确保你可以扩展成上千万的连接,或者好几万的队列,使用JDBC消息存储或者更快的KahaDb消息存储.

      一个完整的扩展配置如下所示

<broker xmlns="http://activemq.apache.org/schema/core"
brokerName="amq-broker"
dataDirectory="${activemq.base}/data">
	<persistenceAdapter>
		<kahaDB directory="${activemq.base}/data" journalMaxFileLength="32mb"/>
	</persistenceAdapter>
	<destinationPolicy>
		<policyMap>
			<policyEntries>
				<policyEntry queue="&gt;" optimizedDispatch="true"/>
			</policyEntries>
		</policyMap>
	</destinationPolicy>
	<systemUsage>
		<systemUsage>
			<memoryUsage>
				<memoryUsage limit="512 mb"/>
			</memoryUsage>
			<storeUsage>
				<storeUsage limit="10 gb" name="foo"/>
			</storeUsage>
			<tempUsage>
				<tempUsage limit="1 gb"/>
			</tempUsage>
		</systemUsage>
	</systemUsage
		<transportConnectors>
			<transportConnector name="openwire" uri="nio://localhost:61616"/>
		</transportConnectors>
</broker>

    

 

         9.3.2 水平扩展

             除了扩展一个代理,可以使用网络来增加用于应用程序的代理的数量.网络自动传送消息给那些包括有兴趣的消费者的代理，你可以配置你的客户端连接代理集群,选择其中一个随机连接（failover://(tcp://broker1:61616,tcp://broker2:61616)?randomize=true）.\

            为了确保在代理上的消息不会成为孤儿,使用dynamicOnly属性,和一个低的预取消息prefetchSize.

            

<networkConnector uri="static://(tcp://remotehost:61617)"  name="bridge"   dynamicOnly="true"  prefetchSize="1" > </networkConnector>

           水平扩展将会之后详细介绍,因为有可能在被消费者消费之前需要经过多个代理.

     9.3.3 流量分配(Traffic partitioning)

           不直接使用网络连接的优势就是减少在代理之间转发消息的开销,