目前公司项目中有用到activemq，两台机器上分别通过共享文件方式搭建了master-slave集群，但两台机器之间并未组建broker cluster，而是在客户端通过软负载的方式随机选择一组提供服务来达到集群扩展的目的。
上面的方案主要问题在于需要通过软负载去实现分布式的负载均衡算法，需要解决一系列问题。
下面的文章就在原有基础上组建broker cluser（activemq自带），基于学习的目的通过一次搭建过程来体验下（毕竟我不是运维人员），下面是效果图：不需要软负载。
为了简单，broker cluster只创建两组，而且全部节点部署在同一台机器上。
节点名称tcp open-write端口管理台端口共享文件 master-a 61616 8161 /Users/iss/data/activemq/activemq-ha-a slave-a 61617 8162 /Users/iss/data/activemq/activemq-ha-a master-b 61618 8163 /Users/iss/data/activemq/activemq-ha-b slave-b 61619 8164 /Users/iss/data/activemq/activemq-ha-b activemq安装由于最新的版本需要jdk1.8，我这里选择的是支持jdk1.7的5.14.3
简单运行，我们只需要修改两个端口即可：这两文件在activemq安装目录的conf中
activemq.xml这里只用到tcp，所以将其它的可以全部删除，修改uri中的端口61616为节点的端口。
我要评论
<transportConnectors>
<!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
jetty.xml
修改下面的port就行，这是activemq的管理系统端口。
<bean id="jettyPort" class="org.apache.activemq.web.WebConsolePort" init-method="start">
<!-- the default port number for the web console -->
<property name="host" value="0.0.0.0"/>
<property name="port" value="8161"/>
</bean>
端口修改好之后，执行下面的脚本即可启动,然后在data目录下查看activemq.log。
bin/activemq startmaster-slave搭建
为了防止activemq单节点出现故障影响提供服务，所以需要有一个备份的节点当主节点出现故障时马上替补上。这里采用共享文件的方式，原理就是让参与高可用的所有节点共用一个数据文件目录，通过文件锁的方式来决定谁是master谁是slave。我们需要做的就是将多个节点的数据目录配置成相同的就行。
环境变量
在bin目录下有个env文件，里面指定了activemq所使用到的各类变量，数据目录路径修改 ACTIVEMQ_DATA：
# Active MQ installation dirs # ACTIVEMQ_HOME="<Installationdir>/" # ACTIVEMQ_BASE="$ACTIVEMQ_HOME" # ACTIVEMQ_CONF="$ACTIVEMQ_BASE/conf" ACTIVEMQ_DATA="/Users/iss/data/activemq/activemq-ha-a/data" # ACTIVEMQ_TMP="$ACTIVEMQ_BASE/tmp"
先启动master，然后再启动slave，如果配置正常，在slave的启动日志中会输出如下日志,表示已经有master锁定，自己将以slave角色运行。
2018-01-01 01:46:56,769 | INFO | Database /Users/iss/data/activemq/activemq-ha-a/data/kahadb/lock is locked by another server. This broker is now in slave mode waiting a lock to be acquired | org.apache.activemq.store.SharedFileLocker | main
当master-a出现故障时系统会自动被slave-a取代。
brocker-cluster搭建上面的高可用只是解决了单点故障问题，同一时间提供服务的只有master一个节点，这显示无法面对数据量的增长需求，所以就需要一种可扩展节点的集群方式来解决面临的问题。让一个broker与其它broker互相通信，我们这里采用静态uri方式，做法还是修改activemq.xml:
master-a与slave-a组成一个broker-a;master-b与slave-b组成一个broker-b，broker-a与broker-b组成broker cluster
broker-a配置修改让其能与broker-b通信
<networkConnectors>
<networkConnector uri="static:(tcp://localhost:61618,tcp://localhost:61619)" duplex="false"/>
</networkConnectors>
broker-b配置修改
让其能与broker-a通信
<networkConnectors>
<networkConnector uri="static:(tcp://localhost:61616,tcp://localhost:61617)" duplex="false"/>
</networkConnectors>
由于本文出于简单演示的目的，只组建了两个broker，它们相互之间的通信配置也很容易。当broker实例比较多时，相互之前的桥接通信的配置还需要仔细研究，待后续补充......
spring-boot示例整个工程结构如下，包含一个生产消息的，一个消费消息的。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
创建activemq启动配置类
brocker url
配置整个集群的url，包含全部master,slave，本文总共是4个。
JmsMessagingTemplate发送消息时支持类，是对JmsTemplate的进一步包装。
JmsListenerContainerFactory
@ComponentScan(basePackages = {"com.jim.framework.activemq"})
@Configuration
public class ActivemqConfiguration {
private static final String BROKER_URL="failover:(tcp://192.168.10.222:61616,tcp://192.168.10.222:61617,tcp://192.168.10.222:61618,tcp://192.168.10.222:61619)";
@Bean
public Queue productActiveMQQueue(){
return new ActiveMQQueue("jim.queue.product");
}
@Bean
public JmsListenerContainerFactory<?> jmsListenerContainerQueue() {
DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
bean.setConnectionFactory(new ActiveMQConnectionFactory(BROKER_URL));
return bean;
}
@Bean
public JmsMessagingTemplate jmsMessagingTemplate(){
return new JmsMessagingTemplate(new ActiveMQConnectionFactory(BROKER_URL));
}
}
定义消息发送接口
public interface ProductSendMessage {
void sendMessage(Object message);
}
实现消息生产者
@Service
public class ProductProducer implements ProductSendMessage {
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
@Autowired
private Queue productActiveMQQueue;
@Override
public void sendMessage(Object message) {
this.jmsMessagingTemplate.convertAndSend(this.productActiveMQQueue,message);
}
}
实现消息消费者
@JmsListener,这个注解即标识监听哪一个消息队列。
@Component
public class ProductConsumer {
@JmsListener(destination = "jim.queue.product",containerFactory = "jmsListenerContainerQueue")
public void receiveQueue(String text) {
System.out.println("Consumer,productId:"+text);
}
}
客户端调用
简单的一个web工程，访问某个链接时发送消息
@RestController
@RequestMapping("/product")
public class ProductController{
@Autowired
private ProductProducer productProducer;
@RequestMapping("/{productId}")
public Long getById(@PathVariable final long productId) {
this.productProducer.sendMessage(productId);
return productId;
}
}
当访问请求后，看看消费方的输出：请求分别转发到了61616以及61618两个master上了，实现了自动负载均衡。
2018-01-01 09:03:43.683 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61616 Consumer,productId:80 2018-01-01 09:03:45.794 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61616 Consumer,productId:80 2018-01-01 09:03:47.745 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61618 Consumer,productId:80 2018-01-01 09:03:49.669 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61616 Consumer,productId:80
模似一个master出现故障，停止master-a后出现这样的日志，显然activemq客户端已经检测到。
2018-01-01 11:25:19.348 WARN 18418 --- [222:61616@55277] o.a.a.t.failover.FailoverTransport : Transport (tcp://192.168.10.222:61616) failed , attempting to automatically reconnect: {}
java.io.EOFException: null
at java.io.DataInputStream.readInt(DataInputStream.java:392) ~[na:1.8.0_121]
at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:268) ~[activemq-client-5.14.5.jar:5.14.5]
at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:240) ~[activemq-client-5.14.5.jar:5.14.5]
at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:232) ~[activemq-client-5.14.5.jar:5.14.5]
at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:215) ~[activemq-client-5.14.5.jar:5.14.5]
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_121]
再次请求测试链接：发现在停止到master-a后，slave-a(61617)已经成功取代原来的master-a(61616)，现在请求已经成功负载到新的master上。
2018-01-01 11:25:19.383 INFO 18418 --- [ActiveMQ Task-3] o.a.a.t.failover.FailoverTransport : Successfully reconnected to tcp://192.168.10.222:61618 2018-01-01 11:26:47.652 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61618 Consumer,productId:80 2018-01-01 11:26:55.408 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61618 Consumer,productId:80 2018-01-01 11:26:57.446 INFO 18418 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport : Successfully connected to tcp://192.168.10.222:61617 Consumer,productId:80本文源码
https://github.com/jiangmin168168/jim-framework/tree/master/jim-framework-activemq
// =_){if(r[p+1].greedy)continue;v=2,k=k.slice(0,P)}m=k}if(y){g&&(f=y[1].length);var w=y.index+f,y=y[0].slice(f),_=w+y.length,S=m.slice(0,w),O=m.slice(_),j=[p,v];S&&j.push(S);var A=new a(i,c?n.tokenize(y,c):y,d,y,h);j.push(A),O&&j.push(O),Array.prototype.splice.apply(r,j)}}}}}return r},hooks:{all:{},add:function(e,t){var a=n.hooks.all;a[e]=a[e]||[],a[e].push(t)},run:function(e,t){var a=n.hooks.all[e];if(a&&a.length)for(var r,l=0;r=a[l++];)r(t)}}},a=n.Token=function(e,t,n,a,r){this.type=e,this.content=t,this.alias=n,this.matchedStr=a||null,this.greedy=!!r};if(a.stringify=function(e,t,r){if("string"==typeof e)return e;if("Array"===n.util.type(e))return e.map(function(n){return a.stringify(n,t,e)}).join("");var l={type:e.type,content:a.stringify(e.content,t,r),tag:"span",classes:["token",e.type],attributes:{},language:t,parent:r};if("comment"==l.type&&(l.attributes.spellcheck="true"),e.alias){var i="Array"===n.util.type(e.alias)?e.alias:[e.alias];Array.prototype.push.apply(l.classes,i)}n.hooks.run("wrap",l);var o="";for(var s in l.attributes)o+=(o?" ":"")+s+'="'+(l.attributes[s]||"")+'"';return""+l.content+""},!_self.document)return _self.addEventListener?(_self.addEventListener("message",function(e){var t=JSON.parse(e.data),a=t.language,r=t.code,l=t.immediateClose;_self.postMessage(n.highlight(r,n.languages[a],a)),l&&_self.close()},!1),_self.Prism):_self.Prism;var r=document.currentScript||[].slice.call(document.getElementsByTagName("script")).pop();return r&&(n.filename=r.src,document.addEventListener&&!r.hasAttribute("data-manual")&&document.addEventListener("DOMContentLoaded",n.highlightAll)),_self.Prism}();"undefined"!=typeof module&&module.exports&&(module.exports=Prism),"undefined"!=typeof global&&(global.Prism=Prism); // ]]> // // ?>?=?|==?|&[&=]?|\|[|=]?|\*=?|\/=?|%=?|\^=?|[?:~])/m,lookbehind:!0}}),Prism.languages.insertBefore("java","function",{annotation:{alias:"punctuation",pattern:/(^|[^.])@\w+/,lookbehind:!0}}); // ]]> // =]+))?)*\s*\/?>/i,inside:{tag:{pattern:/^\/]+/i,inside:{punctuation:/^\/:]+:/}},"attr-value":{pattern:/=(?:('|")[\w\W]*?(\1)|[^\s>]+)/i,inside:{punctuation:/[=>"']/}},punctuation:/\/?>/,"attr-name":{pattern:/[^\s>\/]+/,inside:{namespace:/^[^\s>\/:]+:/}}}},entity:/&#?[\da-z]{1,8};/i},Prism.hooks.add("wrap",function(a){"entity"===a.type&&(a.attributes.title=a.content.replace(/&/,"&"))}),Prism.languages.xml=Prism.languages.markup,Prism.languages.html=Prism.languages.markup,Prism.languages.mathml=Prism.languages.markup,Prism.languages.svg=Prism.languages.markup; // ]]>
如对本文有疑问， 点击进行留言回复！！
网友评论