一、环境准备

1、上传 apache-activemq-5.11.1-bin.tar 和 zookeeper-3.4.5.tar.gz Linux服务器(/usr/local/install 目录下)

zookeeper-3.4.5.tar.gz 上传方式相同。

二、Zookeeper方案

主机IP

消息端口

通信端口

节点目录/usr/local/software 下

192.168.211.128

2181

2888:3888

zookeeper

192.168.211.129

2181

2888:3888

zookeeper

192.168.211.130

2181

2888:3888

zookeeper

1、分别在三台主机上 解压 zookeeper-3.4.5.tar.gz 到 /usr/local/software/ 目录下，解压命令如下：

tar -zxvf zookeeper-3.4.5.tar.gz -C /usr/local/software/

2、到 /usr/local/software 目录下  将 zookeeper-3.4.5 改名为 zookeeper ，命令如下

mv zookeeper-3.4.5  zookeeper (三台主机同样操作)

3、配置环境变量

Vim /etc/profile

修改之后将立即生效 source /etc/profile (三台主机同样操作)

4、Zookeeper配置

进入到 zookeeper conf/目录  将zoo.sample.cfg 修改为 zoo.cfg，如图所示：

编辑 zoo.cfg文件 vim zoo.cfg (三台主机同样操作)

配置说明：

dataDir = /usr/local/software/zookeeper/data  需要在zookeeper 下创建 data目录  如图所示

clientPort:消息端口

server.0=192.168.211.128:2888:3888

server.1=192.168.211.128:2888:3888

server.2=192.168.211.128:2888:3888

最后，进入到 /usr/local/software/zookeeper/data 目录下，创建myid文件  并编辑

在 192.168.211.128 主机上写入0

在 192.168.211.129 主机上写入1

在 192.168.211.130 主机上写入2

至此，zookpeer集群搭建完毕

①、启动zookpeer  zkServer.sh start

②、查看zookeeper的状态  zkServer.sh status

③、连接上 zookeeper 客户端  zkCli.sh

测试Zookeeper集群 启动之后   查看各个 zookpeer 的状态，如图

192.168.211.128：

192.168.211.129：

192.168.211.130：

至此，Zookeeper集群搭建成功

三、ActiveMq集群方案

注：将集群搭建在 192.168.211.130 主机上

主机Ip

集群通信端口

消息端口

控制台端口

节点目录/usr/local/software

192.168.211.130

62621

51511

8161

activemq-cluster/node1

192.168.211.130

62622

51512

8162

activemq-cluster/node2

192.168.211.130

62623

51513

8163

activemq-cluster/node3

1、在 /usr/local/software/目录下  创建 activemq-cluster 文件夹，解压三份  apache-activemq-5.11.1-bin.tar.gz 到

/usr/local/software/activemq-cluster   分别命名为 node1,node2,node2,如图所示：

2、那我们现在已经解压好了三个mq节点也就是node1、node2、node3，下面 我们要做的事情就是更改每个节点不同的配置和端口(由于是在一台机器上实 现集群)。

2.1、修改node1节点的 jetty.xml(/usr/local/software/activemq-cluster/node1/conf)

2.2、修改node2节点的 jetty.xml(/usr/local/software/activemq-cluster/node2/conf)

2.3、修改 node3节点的 jetty.xml (/usr/local/software/activemq-cluster/node3/conf)

3、修改  activemq.xml 文件(/usr/local/software/active-cluster/node1--node2--node3/conf)  一共四处

3.1、修改brokerName 为集群的名字(另外两个节点保持一致)

3.2、先注释掉适配器中的kahadb

3.3、添加新的leveldb配置如下(三个节点都需要修改)

node1节点配置如下：

directory="${activemq.data}/leveldb"

replicas="3"

bind="tcp://0.0.0.0:62621"

zkAddress="192.168.211.128:2181,192.168.211.129:2181,192.168.211.130:2181"

hostname="activemq3"

zkPath="/activemq/leveldb-stores" />

node2节点配置如下：

directory="${activemq.data}/leveldb"

replicas="3"

bind="tcp://0.0.0.0:62622"

zkAddress="192.168.211.128:2181,192.168.211.129:2181,192.168.211.130:2181"

hostname="activemq3"

zkPath="/activemq/leveldb-stores" />

node3节点配置如下：

directory="${activemq.data}/leveldb"

replicas="3"

bind="tcp://0.0.0.0:62623"

zkAddress="192.168.211.128:2181,192.168.211.129:2181,192.168.211.130:2181"

hostname="activemq3"

zkPath="/activemq/leveldb-stores" />

参数说明：

replicas：activeMq集群节点的个数

3.4、修改消息端口

node1 节点修改如下：

node2和node3节点分别修改 51512,51513

至此，activeMq集群搭建完毕

四、测试

在三个节点下启动 activemq集群 ./activemq start

用 zkCli.sh  连接zookeeper 客户端

控制台测试：

192.168.211.130/8161/admin  通过输入用户名admin,密码admin 进入到activeMQ的控制台，也由此可知 8161对外服务  ，访问8162,8163 都不对外服务

代码测试

消费者：

1 public classConsumer {2

3

4

5 public static void main(String[] args) throwsException{6 //第一步：建立ConnectionFactory 对象 ，需要输入用户名、密码、以及要连接的地址，均使用默认即可，默认端口为 "tcp://localhost/61616"

7 ConnectionFactory connectionFactory = newActiveMQConnectionFactory(8 ActiveMQConnectionFactory.DEFAULT_USER,9 ActiveMQConnectionFactory.DEFAULT_PASSWORD,10 "failover:(tcp://192.168.211.130:51511,tcp://192.168.211.130:51512,tcp://192.168.211.130:51513)?Randomize=false");11

12 //第二步：通过 ConnectionFactory 工厂对象我们创建一个 Connection连接，并且通过 Connection的start方法开启连接

13

14 Connection connection =connectionFactory.createConnection();15 connection.start();16

17 //第三步：通过 Connection 对象创建 Session回话(上下文环境对象),用于接收消息，参数配置1是否启用是事务，参数配置2为签收模式，一般我们设置自动签收

18

19 Session session =connection.createSession(Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);20

21 /**

22 * 第四步：通过 Session对象创建Destination对象 ，指的是一个客户端用来指定生产消息目标和消费消息来源的对象，在PTP模式中 Destination被称做Queue(即队列)，23 * 在Pub/Sub 模式中，Destination 被称做Topic即主题，在程序中可以使用多个Queue和Topic24 */

25 Destination destination = session.createQueue("first");26

27 //第五步：我们需要通过Session对象创建消息的发送和接受对象(生产者和消费者)MessageProducer/MessageConsumer

28 MessageConsumer consumer =session.createConsumer(destination);29

30 while(true){31 TextMessage textMessage =(TextMessage) consumer.receive();32

33 //手工去签收消息，另起一个线程(TCP) 去通知我们的MQ 服务去签收

34 textMessage.acknowledge();35 if(textMessage == null){36 break;37 }38 System.out.println("收到的内容：" +textMessage.getText());39 }40

41 if(connection != null){42 connection.close();43 }44

45 }46

47

48 }

生产者：

public static void main(String[] args) throwsException{//第一步：建立ConnectionFactory 对象 ，需要输入用户名、密码、以及要连接的地址，均使用默认即可，默认端口为 "tcp://localhost/61616"

ConnectionFactory connectionFactory = newActiveMQConnectionFactory(

ActiveMQConnectionFactory.DEFAULT_USER,

ActiveMQConnectionFactory.DEFAULT_PASSWORD,"failover:(tcp://192.168.211.130:51511,tcp://192.168.211.130:51512,tcp://192.168.211.130:51513)?Randomize=false");//第二步：通过 ConnectionFactory 工厂对象我们创建一个 Connection连接，并且通过 Connection的start方法开启连接

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(destination);//第六步：我们可以使用MessageProducer的 setDeliveryMode 方法为其设置持久化特性和非持久化特性(DeliveryMode)

producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);//第七步：最后我们使用JMS规范的TextMessage 形式创建数据(通过Session对象) ，并使用 MessageProducer 对象的send方法发送数据

for(int i=0; i<5000; i++){

TextMessage textMessage=session.createTextMessage();

textMessage.setText("我是消息内容" +i);

producer.send(textMessage);

Thread.sleep(1000);

}if(connection != null){

connection.close();

}

}

运行生产者，可以看到控制台如下，此时还没有消息被消费

运行消费者，可以看到有了一个消费者，此时生产者产生的消息在不断的被消费

除此以外，我们也可以从java 控制台看到消息不断的被消费

此时，我们停掉对外服务的 8161服务(即node1 节点)

此时node1节点被停掉，java控制台停顿一会儿，表明没有消息被消费，因为zookeeper 在选举信息的服务来对外服务，选举出来之后，java 控制台的消息继续被消费

我们在刷新网页端的activeMq的控制台，可以看到8161 并不对外提供服务了，

此时对外服务的是 8162端口，即node2节点

到此，我们的ActiveMq集群就搭建完毕了