序

基礎設施

• zookeeper
  程式協調服務框架，用於自動排程多個activemq。在某個activemq服務宕機後，zookeeper會自動排程叢集中其中一個正常的activemq服務成為master主機繼續服務
• activemq
  訊息佇列框架，我們會部署多個activemq服務

規劃

假設我們有三臺伺服器：

• 192.168.0.200
  部署activemq-master、zookeeper
• 192.168.0.201
  部署activemq-slave01
• 192.168.0.202
  部署activemq-slave02

叢集搭建

配置zookeeper

• 解壓zookeeper-3.4.9.tar.gz

$ tar -zxvf zookeeper-3.4.9.tar.gz

• 通過例項建立zookeeper配置檔案

$ cp -R zoo_sample.cfg /home/zookeeper-3.4.9/conf/zoo.cfg

• 配置zookeeper-3.4.9/conf/zoo.cfg，內容如下：

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
 

# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just 
# example sakes.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
 

# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1

關鍵配置說明：
dataDir：zookeeper資料存放目錄
dataLogDir：zookeeper儲存日誌檔案的目錄
clientPort：zookeeper埠
tickTime：zookeeper伺服器之間或客戶端與伺服器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。tickTime以毫秒為單位
initLimit：叢集中的follower伺服器(F)與leader伺服器(L)之間初始連線時能容忍的最多心跳數（tickTime的數量）
syncLimit：叢集中的follower伺服器與leader伺服器之間請求和應答之間能容忍的最多心跳數（tickTime的數量）
server.N=YYY:A:B：伺服器名稱與地址：叢集資訊（伺服器編號，伺服器地址，LF通訊埠，選舉埠），這個配置項的書寫格式比較特殊。其中N表示伺服器編號，YYY表示伺服器的IP地址，A為LF通訊埠，表示該伺服器與叢集中的leader交換的資訊的埠。B為選舉埠，表示選舉新 leader時伺服器間相互通訊的埠（當leader掛掉時，其餘伺服器會相互通訊，選擇出新的leader）。一般來說，叢集中每個伺服器的A埠都 是一樣，每個伺服器的B埠也是一樣。但是當所採用的為偽叢集時，IP地址都一樣，只能時A埠和B埠不一樣

配置activemq

activemq配置，會按照規劃配置3個服務，並統統交由zookeeper進行排程

• 配置核心配置檔案/home/apache-activemq-5.9.1-bin/activemq.xml

<beans
  xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

    <!-- Allows us to use system properties as variables in this configuration file -->
    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <value>file:${activemq.conf}/credentials.properties</value>
        </property>
    </bean>

   <!-- Allows accessing the server log -->
    <bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
          lazy-init="false" scope="singleton"
          init-method="start" destroy-method="stop">
    </bean>

    <!--
        The <broker> element is used to configure the ActiveMQ broker.
    -->
    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-test" dataDirectory="${activemq.data}">

        <destinationPolicy>
            <policyMap>
              <policyEntries>
                <policyEntry topic=">" >
                    <!-- The constantPendingMessageLimitStrategy is used to prevent
                         slow topic consumers to block producers and affect other consumers
                         by limiting the number of messages that are retained
                         For more information, see:

                         http://activemq.apache.org/slow-consumer-handling.html

                    -->
                  <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                  </pendingMessageLimitStrategy>
                </policyEntry>
              </policyEntries>
            </policyMap>
        </destinationPolicy>


        <!--
            The managementContext is used to configure how ActiveMQ is exposed in
            JMX. By default, ActiveMQ uses the MBean server that is started by
            the JVM. For more information, see:

            http://activemq.apache.org/jmx.html
        -->
        <managementContext>
            <managementContext createConnector="false"/>
        </managementContext>

        <!--
            Configure message persistence for the broker. The default persistence
            mechanism is the KahaDB store (identified by the kahaDB tag).
            For more information, see:

            http://activemq.apache.org/persistence.html
        -->
        <persistenceAdapter>
            <!--<kahaDB directory="${activemq.data}/kahadb"/>-->
        <replicatedLevelDB directory="${activemq.data}/kahadb" 
        replicas="3"
        bind="tcp://0.0.0.0:0"
        zkAddress="192.168.0.201:2181"
        zkSessionTimeout="10s"
        hostname="192.168.0.201"
        zkPath="/home/leveldb-stores"
        />
        </persistenceAdapter>


          <!--
            The systemUsage controls the maximum amount of space the broker will
            use before disabling caching and/or slowing down producers. For more information, see:
            http://activemq.apache.org/producer-flow-control.html
          -->
          <systemUsage>
            <systemUsage>
                <memoryUsage>
                    <memoryUsage percentOfJvmHeap="70" />
                </memoryUsage>
                <storeUsage>
                    <storeUsage limit="100 gb"/>
                </storeUsage>
                <tempUsage>
                    <tempUsage limit="50 gb"/>
                </tempUsage>
            </systemUsage>
        </systemUsage>

        <!--
            The transport connectors expose ActiveMQ over a given protocol to
            clients and other brokers. For more information, see:

            http://activemq.apache.org/configuring-transports.html
        -->
        <transportConnectors>
            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
            <transportConnector name="openwire" uri="tcp://0.0.0.0:51511?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
            <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
        </transportConnectors>

        <!-- destroy the spring context on shutdown to stop jetty -->
        <shutdownHooks>
            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
        </shutdownHooks>

    </broker>

    <!--
        Enable web consoles, REST and Ajax APIs and demos
        The web consoles requires by default login, you can disable this in the jetty.xml file

        Take a look at ${ACTIVEMQ_HOME}/conf/jetty.xml for more details
    -->
    <import resource="jetty.xml"/>

</beans>

配置詳細說明：
a、統一設定brokerName
所有activemq服務，必須全部配置brokerName為統一值

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker-test" dataDirectory="${activemq.data}">

b、配置persistenceAdapter
persistenceAdapter設定持久化方式，主要有三種方式：kahaDB（預設方式）、資料庫持久化、levelDB（activemq v5.9.0提供支援）

 <persistenceAdapter>
     <!--<kahaDB directory="${activemq.data}/kahadb"/>-->
    <replicatedLevelDB directory="${activemq.data}/kahadb" 
    replicas="3"
    bind="tcp://0.0.0.0:0"
    zkAddress="192.168.0.201:2181"
    zkSessionTimeout="10s"
    hostname="192.168.0.201"
    zkPath="/home/leveldb-stores"
    />
</persistenceAdapter>

directory：儲存資料的路徑
replicas：叢集中的節點數，以(replicas/2)+1公式表示叢集中至少要正常執行的服務數量】， 3臺叢集那麼允許1臺宕機， 另外兩臺要正常執行
bind：當這個服務節點成為Master， 它會繫結配置好的地址和埠來履行主從複製協議（配置為各activemq服務所在的ip，保證埠不同即可）
zkAddress：zookeeper的ip和port，如果是zookeeper叢集以”,”隔開
zkSessionTimeout：zookeeper回話的超時時間
zkPassword：zookeeper服務平臺的密碼
hostname：activemq服務部署機器的ip
zkPath：zookeeper選舉資訊交換的存貯路徑
sync：認為訊息被消費完成前， 同步資訊所存貯的策略， 如果有多種策略用逗號隔開， ActiveMQ會選擇較強的策略（local_mem, local_disk則肯定選擇存貯在本地硬碟）

c、配置transportConnector訊息埠

<transportConnectors>
    <transportConnector name="openwire" uri="tcp://0.0.0.0:51511?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600"/>
</transportConnectors>

保證各activemq服務的訊息埠不一致即可，比如三臺activemq服務埠分別配置為：51511、51512、51513

• 配置activemq管理平臺訪問埠/home/apache-activemq-5.9.1-bin/jetty.xml

<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
        <property name="host" value="0.0.0.0"/>
        <property name="port" value="8161"/>
</bean>

property屬性port的value值改為你想配置的埠

啟動服務

• 啟動zookeeper
  zookeeper服務一定要優於activemq服務啟動

$ {zookeeper_home}/bin/zkServer.sh start

• 啟動activemq
  依次啟動activemq服務

$ {activemq_home}/bin/active start

管理activemq

zookeeper的策略， 從三臺activemq伺服器選一臺執行， 其他兩臺等待執行， 只是做資料上的主從同步。

客戶端連線activemq

使用Spring管理Bean，配置如下：

 <bean id="activeMQConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="userName" value="admin" />
        <property name="password" value="admin" />
        <property name="brokerURL" value="failover:(tcp://192.168.0.201:51511,tcp://192.168.0.201:51512,tcp://192.168.0.201:51513)?initialReconnectDelay=100" />
</bean>