ActiveMQ(5)-基于ZooKeeper的HA方案

深碍√TFBOYSˉ_ 2022-06-05 11:18 218阅读 0赞

# 1.服务器准备 #

虚拟机环境CentOS6.5  
zookeeper

<table> 
 <thead> 
  <tr> 
   <th>主机名</th> 
   <th>ip</th> 
   <th>zookeeper</th> 
   <th>消息端口</th> 
   <th>集群通信端口</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>hadoop1</td> 
   <td>192.168.72.133</td> 
   <td>1</td> 
   <td>2181</td> 
   <td>2888:3888</td> 
  </tr> 
  <tr> 
   <td>hadoop2</td> 
   <td>192.168.72.129</td> 
   <td>1</td> 
   <td>2181</td> 
   <td>2888:3888</td> 
  </tr> 
  <tr> 
   <td>hadoop3</td> 
   <td>192.168.72.135</td> 
   <td>1</td> 
   <td>2181</td> 
   <td>2888:3888</td> 
  </tr> 
 </tbody> 
</table>

activemq

<table> 
 <thead> 
  <tr> 
   <th>主机名</th> 
   <th>ip</th> 
   <th>activemq</th> 
   <th>消息端口</th> 
   <th>集群通信端口</th> 
   <th>控制台端口</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>hadoop1</td> 
   <td>192.168.72.133</td> 
   <td>1</td> 
   <td>61616</td> 
   <td>62621</td> 
   <td>8161</td> 
  </tr> 
  <tr> 
   <td>hadoop2</td> 
   <td>192.168.72.129</td> 
   <td>1</td> 
   <td>61617</td> 
   <td>62622</td> 
   <td>8162</td> 
  </tr> 
  <tr> 
   <td>hadoop3</td> 
   <td>192.168.72.135</td> 
   <td>1</td> 
   <td>61618</td> 
   <td>62623</td> 
   <td>8163</td> 
  </tr> 
 </tbody> 
</table>

# 2.zookeeper搭建 #

1）zookeeper下载：  
wget [http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz][http_mirror.bit.edu.cn_apache_zookeeper_zookeeper-3.4.6_zookeeper-3.4.6.tar.gz]

2）zookeeper解压  
tar -zxvf zookeeper-3.4.6.tar.gz

3）修改配置文件：  
a)cp zoo\_sample.cfg zoo.cfg将原有的zoo\_sample.cfg 文件改为zoo.cfg  
b)修改zoo.cfg文件，修改dataDir的值为：dataDir=/opt/zookeeper  
c)在zoo.cfg文件的最末尾添加三行配置  
server.1=hadoop1:2888:3888  
server.2=hadoop2:2888:3888  
server.3=hadoop3:2888:3888  
1,2,3分别代表zookeeper所在服务，hadoop1,hadoop2,hadoop3代表zookeeper所在服务器的主机名，其余配置为固定配置

4）文件同步  
将zookeeper配置文件同步到其他机器  
scp /usr/app/zookeeper-3.4.6 root@hadoop2:/usr/app/  
scp /usr/app/zookeeper-3.4.6 root@hadoop3:/usr/app/

5）在三台zookeeper服务器中创建/opt/zookeeper目录，并创建文件myid，文件内容分别1,2,3，即对应的serverid

6）配置zookeeper环境变量  
vim /etc/profile，每台zookeeper服务器添加zookeeper的环境变量  
source /etc/profile使每台zookeeper服务器环境变量生效

7）启动zookeeper  
使用命令：zkServer.sh start

8）查看zookeeper状态  
使用命令：zkServer.sh status

# 3.activemq搭建 #

1）activemq下载  
wget [https://archive.apache.org/dist/activemq/5.14.2/apache-activemq-5.14.2-bin.tar.gz][https_archive.apache.org_dist_activemq_5.14.2_apache-activemq-5.14.2-bin.tar.gz]

2）activemq解压（一般解压到当前文件夹下，如果解压到其他文件，是用-C选项指定具体需要解压到的目录）  
tar -zxvf apache-activemq-5.14.2-bin.tar.gz

3）修改jetty端口  
cd /usr/app/apache-activemq-5.14.2/conf  
vim jetty.xml  
hadoop1,hadoop2,hadoop3三台服务器分别对应的端口为8161,8162,8163  
![这里写图片描述][SouthEast]

4）修改activemq.xml文件  
a）修改brokerName  
![这里写图片描述][SouthEast 1]  
将brokerName改成自己的配置名称，三台机器命名一样

b）注释原持久化kahadb配置

c）添加leveldb持久化配置  
![这里写图片描述][SouthEast 2]  
需要将bind绑定端口改成相应的62621,62622,62623，hostname改为相应的hadoop1，hadoop2，hadoop3

d）修改消息端口  
![这里写图片描述][SouthEast 3]  
端口号相应的改成61616,61617,61618

5）启动activemq  
进入到activemq安装目录  
cd bin  
./activemq start  
查看日志启动是否出现异常，如果没有异常基本启动成功  
![这里写图片描述][SouthEast 4]  
也可以通过zookeeper客户端查看是否注册成功  
![这里写图片描述][SouthEast 5]

配置生效，三个节点都存在  
![这里写图片描述][SouthEast 6]

# 4本地测试 #

sender

package com.mq.activemq_01.helloworld;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    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) throws JMSException {
    
            //1.建立ConnectionFactory工厂对象，连接地址tcp://localhost:61616，在activemq.xml文件中配置
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                                                                                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
    // "tcp://localhost:61616");
                                                                                "failover:(tcp://192.168.72.133:61616,tcp://192.168.72.129:61617,tcp://192.168.72.135:61618)?Randomize=false");
    
            //2.通过ConnectionFactory创建Connection连接，并通过start开启连接
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            //3.通过Connection创建Session会话，参数1表示是否启动事务，参数2表示签收模式封装在Session中
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            //4.通过Session创建Destination目的地，参数生产者和消费者的约束名字，消费者需要知道该名字
            Destination destination = session.createQueue("queue2");
    
            //5.创建消息发送或者接收对象
            MessageProducer messageProducer = session.createProducer(destination);
    
            //6.设置持久化和非持久化对象，这里设置是非持久化
            messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
            //7.通过session创建message，JMS定义了五种正文格式
            for (int i = 1; i <= 5000; i++) {
    
                TextMessage textMessage = session.createTextMessage();
    
                textMessage.setText("我是消息内容，id="+i);
    
                messageProducer.send(textMessage);
    
                System.out.println("生产者：生产了"+textMessage.getText());
                Thread.sleep(1000);
            }
    
    
            //8.释放连接
            if(connection != null){
                connection.close();
            }
    
        }
    }

receiver

package com.mq.activemq_01.helloworld;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class Receiver { 
    
    
        public static void main(String[] args) throws JMSException {
    
    
            //1.建立ConnectionFactory工厂对象，连接地址tcp://localhost:61616，在activemq.xml文件中配置
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnectionFactory.DEFAULT_USER,
                                                                                ActiveMQConnectionFactory.DEFAULT_PASSWORD,
    // "tcp://localhost:61616");
                                                                                "failover:(tcp://192.168.72.133:61616,tcp://192.168.72.129:61617,tcp://192.168.72.135:61618)?Randomize=false");
    
            //2.通过ConnectionFactory创建Connection连接，并通过start开启连接
            Connection connection = connectionFactory.createConnection();
            connection.start();
    
            //3.通过Connection创建Session会话，参数1表示是否启动事务，参数2表示签收模式封装在Session中
            Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    
            //4.通过Session创建Destination目的地，参数生产者和消费者的约束名字，消费者需要知道该名字
            Destination destination = session.createQueue("queue2");
    
            //5.创建消息发送或者接收对象
            MessageConsumer messageConsumer = session.createConsumer(destination);
    
            //6.设置持久化和非持久化对象，这里设置是非持久化
            while (true) {
                TextMessage receive = (TextMessage)messageConsumer.receive();
                if(receive == null){
                    break;
                }
                System.out.println("消费者：收到消息"+receive.getText());
                Thread.sleep(1000);
            }
    
            //7.释放连接
            if(connection != null){
                connection.close();
            }
        }
    }

先启动receiver，再执行sender代码

> INFO | Successfully connected to tcp://192.168.72.133:61616  
> 消费者：收到消息我是消息内容，id=1  
> 消费者：收到消息我是消息内容，id=2  
> 消费者：收到消息我是消息内容，id=3  
> 消费者：收到消息我是消息内容，id=4  
> 消费者：收到消息我是消息内容，id=5  
> 消费者：收到消息我是消息内容，id=6  
> 消费者：收到消息我是消息内容，id=7  
> 消费者：收到消息我是消息内容，id=8  
> 消费者：收到消息我是消息内容，id=9  
> 消费者：收到消息我是消息内容，id=10  
> 消费者：收到消息我是消息内容，id=11  
> 消费者：收到消息我是消息内容，id=12  
> 消费者：收到消息我是消息内容，id=13

![这里写图片描述][SouthEast 7]

# 5故障测试 #

关闭当前master的61616服务  
./activemq stop  
查看执行结果

生产者：

![这里写图片描述][SouthEast 8]

消费者：

![这里写图片描述][SouthEast 9]

自动切换为备用slave

到此，zookeeper+activemq的HF方案架构搭建完毕

[http_mirror.bit.edu.cn_apache_zookeeper_zookeeper-3.4.6_zookeeper-3.4.6.tar.gz]: http://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz
[https_archive.apache.org_dist_activemq_5.14.2_apache-activemq-5.14.2-bin.tar.gz]: https://archive.apache.org/dist/activemq/5.14.2/apache-activemq-5.14.2-bin.tar.gz
[SouthEast]: /images/20220605/61edf342189245eea0999c1a97c99027.png
[SouthEast 1]: /images/20220605/45845391bcbd486e873d7892999d8d9c.png
[SouthEast 2]: /images/20220605/62dd6f9ea9be4ce78a3c0592b7c8aef6.png
[SouthEast 3]: /images/20220605/f3c20963e6734902be2fb2d498ce4369.png
[SouthEast 4]: /images/20220605/4fb0cc4f11344363bf662b7eafc6d05b.png
[SouthEast 5]: /images/20220605/6d3fc2c08fcc4b70ab9c697afef0aa15.png
[SouthEast 6]: /images/20220605/9327ffb10ac94d3cbea91e45e0fb90fc.png
[SouthEast 7]: /images/20220605/6c8835e08ae0433496e1dca875f1e3db.png
[SouthEast 8]: /images/20220605/6622ff3920a347c8aab00fb019c225ef.png
[SouthEast 9]: /images/20220605/a9407a6ace4f4cf8b5efa22f24f3ee0e.png