springboot整合ActiveMQ 2（主備模式，負載均衡）

• 基本使用,https://www.tapme.top/blog/detail/2018-09-05-10-38
• 主備模式,https://www.tapme.top/blog/detail/2018-09-06-10-38

  單個 MQ 節點老是不可靠的，一旦該節點出現故障，MQ 服務就不可用了，勢必會產生較大的損失。這裏記錄 activeMQ 如何開啓主從備份，一旦 master（主節點故障），slave（從節點）當即提供服務，實現原理是運行多個 MQ 使用同一個持久化數據源，這裏以 jdbc 數據源爲例。同一時間只有一個節點（節點 A）可以搶到數據庫的表鎖，其餘節點進入阻塞狀態，一旦 A 發生錯誤崩潰，其餘節點就會從新獲取表鎖，獲取到鎖的節點成爲 master，其餘節點爲 slave，若是節點 A 從新啓動，也將成爲 slave。

  主從備份解決了單節點故障的問題，可是同一時間提供服務的只有一個 master，顯然是不能面對數據量的增加，因此還須要一種橫向拓展的集羣方式來解決面臨的問題。

1、activeMQ 設置

一、平臺版本說明：

• 平臺：windows
• activeMQ 版本：5.9.1，下載地址
• jdk 版本：1.8

二、下載 jdbc 依賴

  下載下面三個依賴包，放到 activeMQ 安裝目錄下的 lib 文件夾中。

mysql 驅動

dhcp 依賴

commons-pool2 依賴

2、主從備份

一、修改 jettty

  首先修改 conf->jetty.xml，這裏是修改 activemq 的 web 管理端口，管理界面帳號密碼默認爲 admin/admin

<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
    <!-- the default port number for the web console -->
    <property name="port" value="8161"/>
</bean>

二、修改 activemq.xml

  而後修改 conf->activemq.xml

• 設置鏈接方式

  默認是下面五種鏈接方式都打開，這裏咱們只要 tcp，把其餘的都註釋掉，而後在這裏設置 activemq 的服務端口，能夠看到每種鏈接方式都對應一個端口。

  <transportConnectors>
    <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
    <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?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>

• 設置 jdbc 數據庫

  mysql 數據庫中建立 activemq 庫，在broker標籤的下面也就是根標籤beans的下一級建立一個 bean 節點，內容以下：

  <bean id="mysql-qs" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
      <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
      <property name="url" value="jdbc:mysql://localhost:3306/activemq?relaxAutoCommit=true"/>
      <property name="username" value="root"/>
      <property name="password" value="123456"/>
      <property name="poolPreparedStatements" value="true"/>
  </bean>

• 設置數據源

  首先修改 broker 節點，設置 name 和 persistent(默認爲 true),也可不作修改，修改後以下：

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

  而後設置持久化方式,使用到咱們以前設置的 mysql-qs

  <persistenceAdapter>
    <!-- <kahaDB directory="${activemq.data}/kahadb"/> -->
    <jdbcPersistenceAdapter dataDirectory="${activemq.base}/activemq-data" dataSource="#mysql-qs"/>
  </persistenceAdapter>

三、啓動

  設置完畢後啓動 activemq（雙擊 bin 中的 acitveMQ.jar)，啓動完成後能夠看到以下日誌信息：

INFO | Using a separate dataSource for locking: org.apache.commons.dbcp2.BasicDataSource@179ece50
 INFO | Attempting to acquire the exclusive lock to become the Master broker
 INFO | Becoming the master on dataSource: org.apache.commons.dbcp2.BasicDataSource@179ece50

​ 接着咱們修改一下 tcp 服務端口，改成 61617，而後從新啓動，日誌信息以下：

INFO | Using a separate dataSource for locking: org.apache.commons.dbcp2.BasicDataSource@179ece50
 INFO | Attempting to acquire the exclusive lock to become the Master broker
 INFO | Failed to acquire lock.  Sleeping for 10000 milli(s) before trying again...
 INFO | Failed to acquire lock.  Sleeping for 10000 milli(s) before trying again...
 INFO | Failed to acquire lock.  Sleeping for 10000 milli(s) before trying again...
 INFO | Failed to acquire lock.  Sleeping for 10000 milli(s) before trying again...

能夠看到從節點一直在嘗試獲取表鎖成爲主節點，這樣一旦主節點失效，從節點可以馬上取代主節點提供服務。這樣咱們便實現了主從備份。

3、負載均衡

  activemq 能夠實現多個 mq 之間進行路由，假設有兩個 mq，分別爲 brokerA 和 brokerB，當一條消息發送到 brokerA 的隊列 test 中，有一個消費者連上了 brokerB，而且想要獲取 test 隊列，brokerA 中的 test 隊列就會路由到 brokerB 上。

   開啓負載均衡須要設置networkConnectors節點，靜態路由配置以下：

<networkConnectors>
  <networkConnector uri="static:failover://(tcp://localhost:61616,tcp://localhost:61617)"           duplex="false"/>
</networkConnectors>

brokerA 和 brokerB 都要設置該配置，以連上對方。

4、測試

一、創建 mq

  組建兩組 broker，每組作主從配置。

• brokerA：
  • 主：設置 web 管理端口 8761,設置 mq 名稱mq,設置數據庫地址爲 activemq，設置 tcp 服務端口 61616，設置負載均衡靜態路由static:failover://(tcp://localhost:61618,tcp://localhost:61619),而後啓動
  • 從：上面的基礎上修改 tcp 服務端口爲 61617,而後啓動
• brokerB:
  • 主：設置 web 管理端口 8762，設置 mq 名稱mq1,設置數據庫地址 activemq1，設置 tcp 服務端口 61618，設置負載均衡靜態路由static:failover://(tcp://localhost:61616,tcp://localhost:61617),而後啓動
  • 從：上面的基礎上修改 tcp 服務端口爲 61619，而後啓動

二、springboot 測試

   沿用上一篇的項目，修改配置文件的 broker-url 爲failover:(tcp://localhost:61616,tcp://localhost:61617,tcp://localhost:61618,tcp://localhost:61619)，而後啓動項目訪問會在控制檯看到以下日誌：

2018-07-31 15:09:25.076  INFO 12780 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61618
1I'm from queue1:hello
2018-07-31 15:09:26.599  INFO 12780 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61618
2I'm from queue1:hello
2018-07-31 15:09:29.002  INFO 12780 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616
1I'm from queue1:hello
2018-07-31 15:09:34.931  INFO 12780 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61618
2I'm from queue1:hello

證實負載均衡成功。

本文原創發佈於：https://www.tapme.top/blog/detail/2018-09-06-10-38