Zookeeper+ActiveMQ集群搭建

本文轉載自查看原文 2018-02-17 11:36 3972 zookeeper ActiveMQ/ Zookeeper

搭建三台虛擬機安裝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

[root@localhost data]# zkServer.sh start
JMX enabled by default
Using config: /usr/local/zookeeper-3.4.5/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[root@localhost 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查看狀態發現有錯誤

[root@localhost 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:QuorumCnxManager@368] - 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:FastLeaderElection@774] - 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:QuorumCnxManager@368] - 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&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>

　　(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&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>

　　(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&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>

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();
        }

    }
}

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 zookeeper+activemq高可用集群搭建 Zookeeper+ActiveMQ搭建集群實現高可用基於zookeeper的activemq的主從集群配置 ActiveMQ集群搭建 ActiveMQ消息隊列集群的搭建 docker搭建zookeeper集群 zookeeper集群搭建 zookeeper 集群搭建基於zookeeper的Swarm集群搭建 ZooKeeper搭建集群