Zookeeper+ActiveMQ叢集搭建
搭建三台虛拟機安裝centos7。要提前安裝好jdk環境
1.環境準備,搭建三台虛拟機ip分别是
- 192.168.192.130
- 192.168.192.131
- 192.168.192.134
Zookeeper環境準備
主機ip | 消息端口 | 通信端口 | 節點目錄/user/local/ |
192.168.192.130 | 2181 | 2888:3888 | zookeeper |
192.168.192.131 | 2181 | 2888:3888 | zookeeper |
192.168.192.134 | 2181 | 2888:3888 | zookeeper |
ActiveMQ環境準備:
主機IP | 叢集通信端口 | 消息端口 | 控制台端口 | 節點目錄/user/local/ |
192.168.192.130 | 62621 | 51511 | 8161 | activemq-cluster/node1 |
192.168.192.131 | 62622 | 51512 | 8162 | activemq-cluster/node2 |
192.168.192.134 | 62623 | 51513 | 8163 | activemq-cluster/node3 |
2.搭建zookeeper環境(三個虛拟機操作一樣)
- 解壓zookeeper-3.4.5.tar.gz 複制到/opt/usr/local下(三台虛拟機都這麼做)
- 配置zookeeper的環境變量。vim/etc/profile(配置完成後要記得source /etc/profile 生效修改)
JAVA_HOME=/usr/lib/java/jdk1.7 ZOOKEEPER_HOME=/usr/local/zookeeper-3.4.5 CLASS_PATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar PATH=$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$PATH export JAVA_HOME CLASS_PATH ZOOKEEPER_HOME PATH
- 修改zookeeper配置檔案名,到zookeeper的conf目錄下将zoo_sample.cfg檔案改名為zoo.cfg
$ mv zoo_sample.cfg zoo.cfg
- 編輯zoo.cfg檔案
$ vim 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=/usr/local/zookeeper-3.4.5/data #日志目錄 dataLogDir=/usr/local/zookeeper-3.4.5/logs # the port at which the clients will connect clientPort=2181 # # 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 #配置三個server server.0=192.168.192.130:2888:3888 server.1=192.168.192.131:2888:3888 server.2=192.168.192.134:2888:3888
- 在zookeeper-3.4.5目錄下建立data檔案夾和logs檔案夾
$ cd /usr/local/zookeeper-3.4.5
$mkdir data
$mkdir logs
- 在三個虛拟機下的data檔案夾下建立三個myid檔案,并且三個檔案裡面分别寫入0,1,2
$ cd /usr/local/zookeeper-3.4.5/data
#192.168.192.130下 vim myid 在檔案裡面寫入0,儲存退出 #192.168.192.131下 vim myid 在檔案裡面寫入1,儲存退出 #192.168.192.134下 vim myid 在檔案裡面寫入2,儲存退出
- 啟動三個zookeeper
$ zkServer.sh start
[[email protected] data]# zkServer.sh start JMX enabled by default Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg Starting zookeeper ... STARTED [[email protected] data]#
- ps:常用指令:
啟動: zkServer.sh start
停止: zkServer.sh stop
重新開機: zkServer.sh restart
檢視服務狀态: zkServer.sh status
- 遇到的問題1:
用zkServer.sh start啟動報錯說zkServer.sh:未找到指令 就用 ./zkServer.sh start 來啟動
- 遇到的問題2:
三台虛拟機啟動後用zkServer.sh status檢視狀态發現有錯誤
[[email protected] conf]# zkServer.sh status JMX enabled by default Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg Error contacting service. It is probably not running.
然後用$zkServer.sh start-foreground指令來啟動檢視啟動日志 發現日志報錯
java.net.NoRouteToHostException: No route to host at java.net.PlainSocketImpl.socketConnect(Native Method) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:339) at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:200) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:182) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) at java.net.Socket.connect(Socket.java:579) at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectOne(QuorumCnxManager.java:354) at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectAll(QuorumCnxManager.java:388) at org.apache.zookeeper.server.quorum.FastLeaderElection.lookForLeader(FastLeaderElection.java:765) at org.apache.zookeeper.server.quorum.QuorumPeer.run(QuorumPeer.java:716) 2016-11-13 18:00:35,602 [myid:0] - WARN [QuorumPeer[myid=0]/0:0:0:0:0:0:0:0:2181:[email protected]] - Cannot open channel to 1 at election address /192.168.192.131:3888
java.net.NoRouteToHostException: No route to host at java.net.PlainSocketImpl.socketConnect(Native Method) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:339) at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:200) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:182) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) at java.net.Socket.connect(Socket.java:579) at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectOne(QuorumCnxManager.java:354) at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectAll(QuorumCnxManager.java:388) at org.apache.zookeeper.server.quorum.FastLeaderElection.lookForLeader(FastLeaderElection.java:765) at org.apache.zookeeper.server.quorum.QuorumPeer.run(QuorumPeer.java:716) 2016-11-13 18:00:36,035 [myid:0] - INFO [QuorumPeer[myid=0]/0:0:0:0:0:0:0:0:2181:[email protected]] - Notification time out: 800 2016-11-13 18:00:36,837 [myid:0] - WARN [QuorumPeer[myid=0]/0:0:0:0:0:0:0:0:2181:[email protected]] - Cannot open channel to 2 at election address /192.168.192.134:3888
初步斷定是防火牆的問題,我們關閉防火牆試試看能不能解決問題
#關閉防火牆
$ systemctl stop firewalld.service
重新啟動zookeeper我們發現可以使用了。問題解決
3、搭建ActiveMQ并和Zookeeper建立叢集
搭建三個ActiveMQ,分在三個伺服器上,每一個伺服器上一個。192.169.192.130是node1,192.168.192.131是node2,192.168.192.134是node3
(1) 解壓 tar -zxvf apache-activemq-5.14.1-bin.tar.gz
(2) 建立檔案夾mkdir /usr/local/activemq-cluster
(3) 将第一步解壓的檔案夾移動到activemq-cluster下
(4) 改名檔案夾 mv apache-activemq-5.14.1 node1 (其他兩台伺服器也如此操作并分别改名為node2,node3)
(5) 修改conf/activemq.xml檔案 vim activemq.xml。修改其中的bind、zkAddress、bhotname、zkPath。
(6) node1第一處修改:brokerName="active-cluster"
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-cluster" dataDirectory="${activemq.data}"> ...... </broker>
(7)node1 第二處修改:
<persistenceAdapter> <!-- <kahaDB directory="${activemq.data}/kahadb"/> --> <replicatedLevelDB directory="${activemq.data}/leveldb" replicas="3" bind="tcp://0.0.0.0:62621" zkAddress="192.168.192.130:2181,192.168.192.131:2181,192.168.192.134:2181" hostname="192.168.192.130" zkPath="/activemq/leveldb-stores"/> </persistenceAdapter>
(8)node1 第三處修改:修改name="openwire"
<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&wireFormat.maxFrameSize=104857600"/> <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors>
(9) node2 第一處修改
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-cluster" dataDirectory="${activemq.data}"> ...... </broker>
(10) node2第二處修改
<persistenceAdapter> <!-- <kahaDB directory="${activemq.data}/kahadb"/> --> <replicatedLevelDB directory="${activemq.data}/leveldb" replicas="3" bind="tcp://0.0.0.0:62622" zkAddress="192.168.192.130:2181,192.168.192.131:2181,192.168.192.134:2181" hostname="192.168.192.131" zkPath="/activemq/leveldb-stores" /> </persistenceAdapter>
(11) node2第三處修改參考
<transportConnectors> <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB --> <transportConnector name="openwire" uri="tcp://0.0.0.0:51512?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors>
(12) node3第一處修改
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="activemq-cluster" dataDirectory="${activemq.data}"> ...... </broker>
(13) node3第二處修改
<persistenceAdapter> <!-- <kahaDB directory="${activemq.data}/kahadb"/> --> <replicatedLevelDB directory="${activemq.data}/leveldb" replicas="3" bind="tcp://0.0.0.0:62623" zkAddress="192.168.192.130:2181,192.168.192.131:2181,192.168.192.134:2181" hostname="192.168.192.134" zkPath="/activemq/leveldb-stores" /> </persistenceAdapter>
(14) node3第三處修改
<transportConnectors> <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB --> <transportConnector name="openwire" uri="tcp://0.0.0.0:51513?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> <transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/> </transportConnectors>
4、叢集搭建完畢,測試一下
- 先啟動zookeeper $ zkServer.sh start(有三個zookeeper,貌似第二個啟動的是leader,用$zkServer.sh status檢視)
- 再啟動activemq $ ./activemq start
- 檢視日志看是否出錯 $ tailf node1/data/activemq.log(最先啟動的是master,其他的都是slaver)
- 在浏覽器中輸入http://192.168.192.130/admin來檢視
5、用代碼來測試一下
package com.lee; 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( ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "failover:(tcp://192.168.192.130:51511,tcp://192.168.192.131:51512,tcp://192.168.192.134: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); System.out.println("我是消息内容" + 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(); } } }
package com.lee; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.MapMessage; import javax.jms.Message; 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( ActiveMQConnectionFactory.DEFAULT_USER, ActiveMQConnectionFactory.DEFAULT_PASSWORD, "failover:(tcp://192.168.192.130:51511,tcp://192.168.192.131:51512,tcp://192.168.192.·34: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(); } } }