activemq 5.9.0新推出的主從實現(xiàn)，基于zookeeper來選舉出一個master，其他節(jié)點自動作為slave實時同步消息。因為有實時同步數(shù)據的slave的存在，master不用擔心數(shù)據丟失，所以leveldb會優(yōu)先采用內存存儲消息，異步同步到磁盤，所以該方式的activemq讀寫性能最好因為選舉機制要超過半數(shù)，所以最少需要3臺節(jié)點，才能實現(xiàn)高可用。如果集群是兩臺則master失效后slave會不起作用，所以集群至少三臺。此種方式僅實現(xiàn)主備功能，避免單點故障，沒有負載均衡功能。
1、環(huán)境準備
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）將zookeeper-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
|
1
|
|
2
3
4
5
6
7
8
9
10
11
|
ticktime=2000initlimit=10synclimit=5datadir=/tmp/zookeeperclientport=2181 server.1=192.168.3.10:2888:3888server.2=192.168.3.11:2888:3888server.3=192.168.3.11:2888:3888 |
（3）創(chuàng)建/tmp/zookeeper目錄
在該目錄下創(chuàng)建名為myid的文件，內容為1（這個值隨server而改變）
（4）將server1上的/home/wzh/zk/zookeeper-3.5.2-alpha文件夾復制到server2，server3,然后創(chuàng)建/tmp/zookeeper目錄
在該目錄下創(chuàng)建名為myid的文件，內容為2
（5）啟動zookeeper
[192.168.3.10]
|
1
|
|
2
3
4
|
wzh@hd-master:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkserver.sh startzookeeper jmx enabled by defaultusing config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfgstarting zookeeper ... started |
[192.168.3.11]
|
1
|
|
2
3
4
|
wzh@hd-slave1:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkserver.sh startzookeeper jmx enabled by defaultusing config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfgstarting zookeeper ... started |
[192.168.3.12]
|
1
|
|
2
3
4
|
wzh@hd-slave2:~/zk/zookeeper-3.5.2-alpha/bin$ ./zkserver.sh startzookeeper jmx enabled by defaultusing config: /home/wzh/zk/zookeeper-3.5.2-alpha/bin/../conf/zoo.cfgstarting zookeeper ... started |
3、搭建activemq集群
（1）將apache-activemq-5.13.0-bin.tar.gz解壓到/home/wzh/amq
（2）修改activemq.xml配置文件
【1】將broker節(jié)點的brokername設置為wzhamq
【2】將persistenceadapter的持久化方式選用replicatedleveldb，將kahadb方式注釋掉
|
1
|
|
2
3
4
5
6
7
8
9
10
11
12
13
|
<persistenceadapter> <!-- <kahadb directory="${activemq.data}/kahadb"/> --> <replicatedleveldb 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"/> </persistenceadapter> |
將apache-activemq-5.13.復制到11，12機器
wzh@hd-master:~/amq$ scp -r apache-activemq-5.13.0/ [email protected]:/tmp
修改配置文件中的hostname="192.168.3.11"
修改配置文件中的hostname="192.168.3.12"
（3）啟動activemq
|
1
|
|
2
3
4
5
|
wzh@hd-master:~/amq$ ./apache-activemq-5.13.0/bin/activemq statusinfo: 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文件中增加
|
1
|
|
2
3
4
5
|
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中增加以下配置
|
1
|
|
2
3
4
5
|
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=truespring.activemq.pool.enabled=falsespring.activemq.user=adminspring.activemq.password=admin |
（3）jms消息發(fā)送
|
1
|
|
2
3
4
5
6
7
8
9
10
|
@servicepublic class producer { @autowired private jmsmessagingtemplate jmstemplate; public void sendmessage(destination destination, final string message){ jmstemplate.convertandsend(destination, message); }} |
（4）jms消息接收
|
1
|
|
2
3
4
5
6
7
8
|
@componentpublic class consumer { @jmslistener(destination = "test.queue") public void receivequeue(string text){ system.out.println("consumer收到的報文為:"+text); }} |
（5）測試
|
1
|
|
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
@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<string, object> say(@pathvariable string msg, @pathvariable string name){ map<string, object> map = new hashmap<>(); map.put("msg", msg); map.put("name", name); producer.sendmessage(destination, msg); return map; }} |
（6）進入activemq管理控制臺創(chuàng)建一個消息隊列
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）目前系統(tǒng)連接的是10，如果此時將10集群down掉，系統(tǒng)會理解選擇一臺slave作為master提供服務，從而啟動案例主備的效果。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
