Zookeeper + ActiveMQ 叢集整合
一、使用ZooKeeper實現的Master-Slave實現方式，是對ActiveMQ進行高可用的一種有效的解決方案，高可用的原理：使用ZooKeeper（叢集）註冊所有的ActiveMQ Broker。只有其中的一個Broker可以對外提供服務（也就是Master節點），其他的Broker處於待機狀態，被視為Slave。如果Master因故障而不能提供服務，則利用ZooKeeper的內部選舉機制會從Slave中選舉出一個Broker充當Master節點，繼續對外提供服務。
二：部署方案，ActiveMQ叢集環境準備：
（1）首先我們下載apache-activemq-5.11.1-bin.tar.gz，到我們的一臺主節點上去，然後我們在（192.168.100.142一個節點上實現叢集即可）
（2）Zookeeper方案
主機IP 訊息埠 通訊埠 節點目錄/usr/local/下
192.168.100.142 2181 2888:3888 zookeeper
192.168.100.143 2181 2888:3888 zookeeper
192.168.100.144 2181 2888:3888 zookeeper
（3）ActiveMQ方案
主機IP 叢集通訊埠 訊息埠 控制檯埠 節點目錄/usr/local/下
192.168.100.142 62621 51511 8161 activemq/node1/
192.168.100.142 62622 51512 8162 activemq/node2/
192.168.100.142 62623 51513 8163 activemq/node3/
2：首先搭建zookeeper環境
1）在192.168.100.142節點下，建立/usr/local/zookeeper
資料夾，解壓zookeeper-3.4.5.tar.gz檔案，然後對解壓好的檔案改名，操作如下：
1 .1命令：mkdir /usr/local/zookeeper
1.2命令：tar -zxvf zookeeper-3.4.5.tar.gz
1.3配置環境變數命令：vim /etc/profile
export ZOOKEEPER_HOME=/usr/local/zookeeper/zookeeper-3.4.5
export PATH=$ZOOKEEPER_HOME/bin:$PATH1.4:使環境變數生效：source /etc/profile
1.5 cd /usr/local/zookeeper/zookeeper-3.4.5/conf
複製一份zoo_sample.cfg改名為zoo.cfg
vim zoo.cig修改
dataDir=/usr/local/zookeeper/zookeeper-3.4.5/data 增加
server.0=192.168.100.142:2888:3888
server.1=192.168.100.143:2888:3888
server.2=192.168.100.144:2888:38881.6:啟動zookeeper:zkServer.sh start
1.7、通過zkServer.sh status檢視狀態
3：繼續搭建activemq環境
（1）在192.168.100.142節點下，建立/usr/local/activemq
資料夾，解壓apache-activemq-5.11.1-bin.tar.gz檔案，然後對解壓好的檔案改名，操作如下：
1 命令：mkdir /usr/local/activemq
3 命令：tar -zxvf apache-activemq-5.11.1-bin.tar.gz
4 命令：cd /usr/local/activemq/
5 命令：mv apache-activemq-5.11.1/ node1 如此操作，再次反覆解壓apache-activemq-5.11.1-bin.tar.gz檔案到/usr/local/activemq/下，建立node2和node3資料夾，如下：
（2）那我們現在已經解壓好了三個mq節點也就是node1、node2、node3，下面我們要做的事情就是更改每個節點不同的配置和埠（由於是在一臺機器上實現叢集）。
1 修改控制檯埠（預設為8161）,在mq安裝路徑下的conf/jetty.xml進
行修改即可。（三個節點都要修改，並且埠都不同）
命令：vim /usr/local/activemq/node1/conf/jetty.xml
三個節點都需要修改為8161、8162、8163
2 叢集配置檔案修改：我們在mq安裝路徑下的conf/activemq.xml進行修改其中的持久化介面卡，修改其中的bind、zkAddress、hostname、zkPath。
然後也需要修改mq的brokerName，並且每個節點名稱都必須相同。
命令：vim /usr/local/activemq-cluster/node1/conf/activemq.xml
第一處修改：brokerName=”activemq-cluster”（三個節點都需要修改）
第二處修改：新增新的leveldb配置如下（三個節點都需要修改）：
Node1:
<persistenceAdapter>
<!--kahaDB directory="${activemq.data}/kahadb"/ -->
<replicatedLevelDB
directory="${activemq.data}/leveldb"
replicas="3"
bind="tcp://0.0.0.0:62621"
zkAddress="192.168.100.142:2181,192.168.100.143:2181,192.168.100.144:2181"
hostname="localhost.localdomain"
zkPath="/activemq/leveldb-stores"
/>
</persistenceAdapter>
Node2:
<persistenceAdapter>
<!--kahaDB directory="${activemq.data}/kahadb"/ -->
<replicatedLevelDB
directory="${activemq.data}/leveldb"
replicas="3"
bind="tcp://0.0.0.0:62622"
zkAddress="192.168.100.142:2181,192.168.100.143:2181,192.168.100.144:2181"
hostname="localhost.localdomain"
zkPath="/activemq/leveldb-stores"
/>
</persistenceAdapter>
Node3:
<persistenceAdapter>
<!--kahaDB directory="${activemq.data}/kahadb"/ -->
<replicatedLevelDB
directory="${activemq.data}/leveldb"
replicas="3"
bind="tcp://0.0.0.0:62623"
zkAddress="192.168.100.142:2181,192.168.100.143:2181,192.168.100.144:2181"
hostname="localhost.localdomain"
zkPath="/activemq/leveldb-stores"
/>
</persistenceAdapter>
命令：vim /usr/local/activemq-cluster/node1/conf/activemq.xml
修改這個檔案的通訊埠號，三個節點都需要修改（51511,51512,51513）
第一步：啟動zookeeper叢集，命令：zkServer.sh start
第二步：啟動mq叢集：順序啟動mq:命令如下：
/usr/local/activemq/node1/bin/activemq start（關閉stop）
/usr/local/activemq/node2/bin/activemq start（關閉stop）
/usr/local/activemq/node3/bin/activemq start（關閉stop）
第三步：檢視日誌資訊：
tail -f /usr/local/activemq/node1/data/activemq.log
tail -f /usr/local/activemq/node2/data/activemq.log
tail -f /usr/local/activemq/node3/data/activemq.log
如果不報錯，我們的叢集啟動成功，可以使用控制檯檢視！
第四步：叢集的brokerUrl配置進行修改即可：
failover:(tcp://192.168.100.142:51511,tcp://192.168.100.143:51512,tcp://192.168.1.144:51513)?Randomize=false
/usr/local/activemq/node1/bin/activemq stop
/usr/local/activemq/node2/bin/activemq stop
/usr/local/activemq/node3/bin/activemq stop
測試程式碼：生產者：Sender
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Sender {
public static void main(String[] args) {
try {
// 第一步：建立ConnectionFactory工廠物件，需要填入使用者名稱、密碼、以及要連線的地址，均使用預設即可，預設埠為"tcp://localhost:61616"
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("xjd", "xjd",
"failover:(tcp://192.168.100.142:51511,tcp://192.168.100.143:51512,tcp://192.168.100.144:51513)?Randomize=false");
// 第二步：通過ConnectionFactory工廠物件我們建立一個Connection連線，並且呼叫Connection的start方法開啟連線，Connection預設是關閉的。
Connection connection = connectionFactory.createConnection();
connection.start();
// 第三步：通過Connection物件建立Session會話（上下文環境物件），用於接收訊息，引數配置1為是否啟用是事務，引數配置2為簽收模式，一般我們設定自動簽收。
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 第四步：通過Session建立Destination物件，指的是一個客戶端用來指定生產訊息目標和消費訊息來源的物件，在PTP模式中，Destination被稱作Queue即佇列；在Pub/Sub模式，Destination被稱作Topic即主題。在程式中可以使用多個Queue和Topic。
Destination destination = session.createQueue("first");
// 第五步：我們需要通過Session物件建立訊息的傳送和接收物件（生產者和消費者）MessageProducer/MessageConsumer。
MessageProducer producer = session.createProducer(null);
// 第六步：我們可以使用MessageProducer的setDeliveryMode方法為其設定持久化特性和非持久化特性（DeliveryMode），我們稍後詳細介紹。
// producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 第七步：最後我們使用JMS規範的TextMessage形式建立資料（通過Session物件），並用MessageProducer的send方法傳送資料。同理客戶端使用receive方法進行接收資料。最後不要忘記關閉Connection連線。
for (int i = 0; i < 500000; i++) {
TextMessage msg = session.createTextMessage("我是訊息內容" + i);
// 第一個引數目標地址
// 第二個引數 具體的資料資訊
// 第三個引數 傳送資料的模式
// 第四個引數 優先順序
// 第五個引數 訊息的過期時間
producer.send(destination, msg, DeliveryMode.NON_PERSISTENT, 0, 1000L);
System.out.println("傳送訊息：" + msg.getText());
Thread.sleep(1000);
}
if (connection != null) {
connection.close();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
消費者：
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
public class Receiver {
public static void main(String[] args) {
try {
// 第一步：建立ConnectionFactory工廠物件，需要填入使用者名稱、密碼、以及要連線的地址，均使用預設即可，預設埠為"tcp://localhost:61616"
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("xjd", "xjd",
"failover:(tcp://192.168.100.142:51511,tcp://192.168.100.143:51512,tcp://192.168.100.144:51513)?Randomize=false");
// 第二步：通過ConnectionFactory工廠物件我們建立一個Connection連線，並且呼叫Connection的start方法開啟連線，Connection預設是關閉的。
Connection connection = connectionFactory.createConnection();
connection.start();
// 第三步：通過Connection物件建立Session會話（上下文環境物件），用於接收訊息，引數配置1為是否啟用是事務，引數配置2為簽收模式，一般我們設定自動簽收。
Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 第四步：通過Session建立Destination物件，指的是一個客戶端用來指定生產訊息目標和消費訊息來源的物件，在PTP模式中，Destination被稱作Queue即佇列；在Pub/Sub模式，Destination被稱作Topic即主題。在程式中可以使用多個Queue和Topic。
Destination destination = session.createQueue("first");
// 第五步：通過Session建立MessageConsumer
MessageConsumer consumer = session.createConsumer(destination);
while (true) {
TextMessage msg = (TextMessage) consumer.receive();
if (msg == null)
break;
System.out.println("收到的內容：" + msg.getText());
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
開啟瀏覽器輸入192.168.100.142:8161/admin檢視佇列的消費情況，然後stop192.168.100.142的activemq服務，就可以完成故障轉。移到另外2臺機器的任意一臺。