ActiveMQ(13):ActiveMQ的集群
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ActiveMQ(13):ActiveMQ的集群相关的知识,希望对你有一定的参考价值。
一、简介
1.1 消费者集群(Queue consumer clusters)
ActiveMQ支持Consumer对消息高可靠性的负载平衡消费,如果一个Consumer死掉,该消息会转发到其它的Consumer消费的Queue上。
如果一个Consumer获得消息比其它Consumer快,那么他将获得更多的消息。
因此推荐ActiveMQ的Broker和Client使用failover://transport的方式来配置链接。
1.2 Broker clusters
大部情况下是使用一系列的Broker和Client链接到一起。如果一个Broker死掉了,Client可以自动链接到其它Broker上。
实现以上行为需要用failover协议作为Client。
如果启动了多个Broker,Client可以使用static discover或者 Dynamic discovery容易的从一个broker到另一个broker直接链接。
这样当一个broker上没有Consumer的话,那么它的消息不会被消费的,然而该broker会通过存储和转发的策略来把该消息发到其它broker上。
特别注意:
ActiveMQ默认的两个broker,static链接后是单方向的,broker-A可以访问消费broker-B的消息,如果要支持双向通信,需要在
netWorkConnector配置的时候,设置duplex=true 就可以了。
操作,服务端搭建静态网络连接与消息回流
消息者1:
public void test1() throws Exception { ConnectionFactory cf = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.175.13:61676"); Connection connection = cf.createConnection(); connection.start(); final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("my-queue"); for (int i = 0; i < 1; i++) { MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { TextMessage m = (TextMessage) message; try { System.out.println("===收到11111111:" + m.getText()); session.commit(); } catch (JMSException e) { e.printStackTrace(); } } }); } }
消息者2:
public void test1() throws Exception { ConnectionFactory cf = new ActiveMQConnectionFactory("liuy","123456","tcp://192.168.175.13:61616"); Connection connection = cf.createConnection(); connection.start(); final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("my-queue"); for (int i = 0; i < 1; i++) { MessageConsumer consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { TextMessage m = (TextMessage) message; try { System.out.println("===收到222222222:" + m.getText()); session.commit(); } catch (JMSException e) { e.printStackTrace(); } } }); } }
生产者:
public void test1() throws Exception { ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("liuy","123456","failover:(tcp://192.168.175.13:61616,tcp://192.168.175.13:61676)"); Connection connection = connectionFactory.createConnection(); connection.start(); Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("my-queue"); MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 30; i++) { TextMessage message = session.createTextMessage("message--" + i); Thread.sleep(1000); producer.send(message); } session.commit(); session.close(); connection.close(); }
效果:
二、主从节点的集群(Master Slave)
在5.9的版本里面,废除了Pure Master Slave的方式,目前支持:
1:Shared File System Master Slave:
基于共享储存的Master-Slave:多个broker实例使用一个存储文件,谁拿到文件锁就是master,其他处于待启动状态,如果master挂掉了,
某个抢到文件锁的slave变成master
2:JDBC Master Slave:基于JDBC的Master-Slave:使用同一个数据库,拿到LOCK表的写锁的broker成为master
3:Replicated LevelDB Store:基于ZooKeeper复制LevelDB存储的Master-Slave机制,这个是5.9新加的
具体的可以到官方察看: http://activemq.apache.org/masterslave.html
注意:这里可以不要静态连接与回流了
2.1 JDBC Master Slave的方式
2.1.1 简介
利用数据库作为数据源,采用Master/Slave模式,其中在启动的时候Master首先获得独有锁,其它Slaves Broker则等待获取独有锁。
推荐客户端使用Failover来链接Brokers。
具体如下图所示:
Master失败
如果Master失败,则它释放独有锁,其他Slaver则获取独有锁,其它Slaver立即获得独有锁后此时它将变成Master,并且启动所有的传输链接。
同时,Client将停止链接之前的Master和将会轮询链接到其他可以利用的Broker即新Master。如上中图所示
Master重启
任何时候去启动新的Broker,即作为新的Slave来加入集群,如上右图所示
2.1.2 JDBC Master Slave的配置
使用<jdbcPersistenceAdapter/>来配置消息的持久化,自动就会使用JDBC Master Slave的方式。
参考:ActiveMQ消息存储持久化 里的jdbc
去掉静态连接:参考ActiveMQ的静态网络链接
去掉回流:参考集群下的消息回流功能
注意:在配置JDBC时,注意配置useDatabaseLock="true",如下
<jdbcPersistenceAdapter dataSource="#mysql-ds" useDatabaseLock="true" />
必需设置,不然在保存数据时会报数据主键重复异常
2.1.3 测试
生产30个消息:
public void test1() throws Exception { ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("liuy","123456","failover:(tcp://192.168.175.13:61616,tcp://192.168.175.13:61676)"); Connection connection = connectionFactory.createConnection(); connection.start(); Session session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("my-queue"); MessageProducer producer = session.createProducer(destination); for (int i = 0; i < 30; i++) { TextMessage message = session.createTextMessage("message--" + i); Thread.sleep(1000); producer.send(message); } session.commit(); session.close(); connection.close(); }
消费3个消息:
public void test1() throws Exception { ConnectionFactory cf = new ActiveMQConnectionFactory("liuy","123456","failover:(tcp://192.168.175.13:61616,tcp://192.168.175.13:61676)"); Connection connection = cf.createConnection(); connection.start(); final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue("my-queue"); MessageConsumer consumer = session.createConsumer(destination); int i=0; while(i<3) { i++; TextMessage message = (TextMessage) consumer.receive(); session.commit(); System.out.println("收到消 息:" + message.getText()); } session.close(); connection.close(); }
然后关闭61616这台mq,进入到61676界面:
本文出自 “我爱大金子” 博客,请务必保留此出处http://1754966750.blog.51cto.com/7455444/1919608
以上是关于ActiveMQ(13):ActiveMQ的集群的主要内容,如果未能解决你的问题,请参考以下文章