ActiveMQ基于zookeeper的主从（levelDB Master/Slave）搭建

ActiveMQ基于zookeeper的主从（levelDB Master/Slave）搭建

ActiveMQ 5.9.0新推出的主从实现，基于zookeeper来选举出一个master，其他节点自动作为slave实时同步消息。因为有实时同步数据的slave的存在，master不用担心数据丢失，所以leveldb会优先采用内存存储消息，异步同步到磁盘，所以该方式的activeMQ读写性能最好因为选举机制要超过半数，所以最少需要3台节点，才能实现高可用。如果集群是两台则master失效后slave会不起作用，所以集群至少三台。此种方式仅实现主备功能，避免单点故障，没有负载均衡功能。

1、环境准备

IP

192.168.3.10    server1

192.168.3.11    server2

192.168.3.12　server3

安装软件信息：

apache-activemq-5.13.0-bin.tar.gz

zookeeper-3.5.2-alpha.tar.gz

ZooInspector.zip

2、搭建Zookeeper集群

（1）将zoolekjJDkeeper-3.5.2-alpha.tar.gz文件解压到/home/wzh/zk目录；

（2）将zoo_sample.cfg复制一份为 zoo.cfg，并修改其配置信息

wzh@hd-master:~/zk/zookeeper-3.5.2-alpha/conf$ cp zoo_sample.cfg zoo.cfg

wzh@hd-master:~/zk/zookeeper-3.5.2-alpha/conf$vim zoo.cfg

tickTime=2000

initLimit=10

syncLimit=5

dataDir=/tmp/zookeeper

clientPort=2181

server.1=192.168.3.10:2888:3888

server.2=192.168.3.11:2888:3888

server.3=192.168.3.11:2888:3888

（3）创建/tmp/zookeeper目录

在该目录下创建名为myid的文件，内容为1（这个值随server而改变）

（4）将server1上的/home/wzh/zk/zookeeper-3.5.2-alpha文件夹复制到server2，server3,然后创建/tmp/zookeeper目录

在该目录下创建名为myid的文件，内容为2

（5）启动zookeeper

[192.168.3.10]

wzh@hd-master:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkServer.sh start

ZooKeeper JMX enabled by default

Using config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

[192.168.3.11]

wzh@hd-slave1:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkServer.sh start

ZooKeeper JMX enabled by default

Using config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

[192.168.3.12]

wzh@hd-slave2:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkServer.sh start

ZooKeeper JMX enabled by default

Using config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfg

Starting zookeeper ... STARTED

3、搭建ActiveMQ集群

（1）将apache-activemq-5.13.0-bin.tar.gz解压到/home/wzh/amq

（2）修改activemq.xml配置文件

【1】将broker节点的brokerName设置为wzhamq

复制代码 代码如下:

【2】将persistenceAdapter的持久化方式选用replicatedLevelDB，将kahaDB方式注释掉

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

replicas="3"

bind="tcp://0.0.0.0:0"

zkAddress="192.168.3.10:2181,192.168.3.11:2181"

hostname="192.168.3.10"

sync="local_disk"

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

将apache-activemq-5.13.复制到11，12机器

wzh@hd-master:~/amq$ scp -r apache-activemq-5.13.0/ wzh@192.168.3.11:/tmp

修改配置文件中的hostname="192.168.3.11"

修改配置文件中的hostname="192.168.3.12"

（3）启动ActiveMQ

wzh@hd-master:~/amq$ ./apache-activemq-5.13.0/bin/activemq status

INFO: Loading '/home/wzh/amq/apache-activemq-5.13.0//bin/env'

INFO: Using java '/opt/java/jdk1.8.0_91/bin/java'

ActiveMQ is running (pid '2031')

wzh@hd-master:~/amq$

依次启动192.168.3.11，192.168.3.12机器

4：集群管理

（1）通过使用ZooInspector工具查看zookeeper集群情况

（2）http://192.168.3.10:8161/admin/ 默认用户名与口令为admin登录ActiveMQ管理端

5、通过Spring-boot操作ActiveMQ JMS

（1）通过gradle构建Spring-boot应用，在 gradle文件中增加

dependencies {

compile('org.springframework.boot:spring-boot-starter-activemq')

compile('org.springframework.boot:spring-boot-starter-web')

testCompile('org.springframework.boot:spring-boot-starter-test')

}

（2）application中增加以下配置

spring.activemq.broker-url=failover:(tcp://192.168.3.10:61616,tcp://192.168.3.11:61616,tcp://192.168.3.12:61616)

spring.activemq.in-memory=true

spring.activemq.pool.enabled=false

spring.activemq.user=admin

spring.activemq.password=admin

（3）JMS消息发送

@Service

public class Producer {

@Autowired

private JmsMessagingTemplate jmsTemplate;

public void sendMessage(Destination destination, final String message){

jmsTemplate.convertAndSend(destination, message);

}

}

（4）JMS消息接收

@Component

public class Consumer {

@JmsListener(destination = "test.queue")

public void receiveQueue(String text){

System.out.println("Consumer收到的报文为:"+text);

}

}

（5）测试

@RestController

@RequestMapping(

value = "/test",

headers = "Accept=application/json",

produces = "application/json;charset=utf-8"

)

public class TestCtrl {

@Autowired

Producer producer;

Destination destination = new ActiveMQQueue("test.queue");

@RequestMapping(

value = "/say/{msg}/to/{name}",

method = RequestMethod.GET

)

public Map say(@PathVariable String msg, @PathVariable String name){

Map map = new HashMap<>();

map.put("msg", msg);

map.put("name", name);

producer.sendMessage(destination, msg);

return map;

}

}

（6）进入ActiveMQ管理控制台创建一个消息队列

test.queue

（7）通过POSTMAN进行测试

2017-08-03 08:10:44.928 INFO 12820 --- [ActiveMQ Task-3] o.a.a.t.failover.FailoverTransport : Successfully reconnected to tcp://192.168.3.10:61616

2017-08-03 08:11:08.854 INFO 12820 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.3.10:61616

Consumer收到的报文为:hello

2017-08-03 08:43:39.464 INFO 12820 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.3.10:61616

Consumer收到的报文为:hello

（8）目前系统连接的是10，如果此时将10集群Down掉，系统会理解选择一台slave作为master提供服务，从而启动案例主备的效果。

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

replicas="3"

bind="tcp://0.0.0.0:0"

zkAddress="192.168.3.10:2181,192.168.3.11:2181"

hostname="192.168.3.10"

sync="local_disk"

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

将apache-activemq-5.13.复制到11，12机器

wzh@hd-master:~/amq$ scp -r apache-activemq-5.13.0/ wzh@192.168.3.11:/tmp

修改配置文件中的hostname="192.168.3.11"

修改配置文件中的hostname="192.168.3.12"

（3）启动ActiveMQ

wzh@hd-master:~/amq$ ./apache-activemq-5.13.0/bin/activemq status

INFO: Loading '/home/wzh/amq/apache-activemq-5.13.0//bin/env'

INFO: Using java '/opt/java/jdk1.8.0_91/bin/java'

ActiveMQ is running (pid '2031')

wzh@hd-master:~/amq$

依次启动192.168.3.11，192.168.3.12机器

4：集群管理

（1）通过使用ZooInspector工具查看zookeeper集群情况

（2）http://192.168.3.10:8161/admin/ 默认用户名与口令为admin登录ActiveMQ管理端

5、通过Spring-boot操作ActiveMQ JMS

（1）通过gradle构建Spring-boot应用，在 gradle文件中增加

dependencies {

compile('org.springframework.boot:spring-boot-starter-activemq')

compile('org.springframework.boot:spring-boot-starter-web')

testCompile('org.springframework.boot:spring-boot-starter-test')

}

（2）application中增加以下配置

spring.activemq.broker-url=failover:(tcp://192.168.3.10:61616,tcp://192.168.3.11:61616,tcp://192.168.3.12:61616)

spring.activemq.in-memory=true

spring.activemq.pool.enabled=false

spring.activemq.user=admin

spring.activemq.password=admin

（3）JMS消息发送

@Service

public class Producer {

@Autowired

private JmsMessagingTemplate jmsTemplate;

public void sendMessage(Destination destination, final String message){

jmsTemplate.convertAndSend(destination, message);

}

}

（4）JMS消息接收

@Component

public class Consumer {

@JmsListener(destination = "test.queue")

public void receiveQueue(String text){

System.out.println("Consumer收到的报文为:"+text);

}

}

（5）测试

@RestController

@RequestMapping(

value = "/test",

headers = "Accept=application/json",

produces = "application/json;charset=utf-8"

)

public class TestCtrl {

@Autowired

Producer producer;

Destination destination = new ActiveMQQueue("test.queue");

@RequestMapping(

value = "/say/{msg}/to/{name}",

method = RequestMethod.GET

)

public Map say(@PathVariable String msg, @PathVariable String name){

Map map = new HashMap<>();

map.put("msg", msg);

map.put("name", name);

producer.sendMessage(destination, msg);

return map;

}

}

（6）进入ActiveMQ管理控制台创建一个消息队列

test.queue

（7）通过POSTMAN进行测试

2017-08-03 08:10:44.928 INFO 12820 --- [ActiveMQ Task-3] o.a.a.t.failover.FailoverTransport : Successfully reconnected to tcp://192.168.3.10:61616

2017-08-03 08:11:08.854 INFO 12820 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.3.10:61616

Consumer收到的报文为:hello

2017-08-03 08:43:39.464 INFO 12820 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.3.10:61616

Consumer收到的报文为:hello

（8）目前系统连接的是10，如果此时将10集群Down掉，系统会理解选择一台slave作为master提供服务，从而启动案例主备的效果。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。