hq82001
- 浏览: 273742 次
- 性别:
- 来自: 南宁
-
文章分类
- 全部博客 (249)
- Memcached (6)
- 操作系统-Linux (18)
- 操作系统-Windows (2)
- JAVA-多线程 (6)
- Java-Web (46)
- Eclipse (19)
- Struts-Spring-Hibernate (25)
- 消息队列 (10)
- JBPM (3)
- 全文检索-Solr (2)
- WebService (3)
- 数据库-Oracle (9)
- 数据库-MySQL (14)
- 数据库-MS SQL (2)
- ESB(企业服务总线) (1)
- 前端技术 (9)
- 互联网安全技术 (1)
- 日志log4j (3)
- 分布式框架 (11)
- 版本控制-SVN (3)
- 版本控制-Git (5)
- Maven (11)
- 大数据-Hadoop (3)
- PHP (1)
- JAVA注解 (2)
- 系统架构 (1)
- 会话-权限-JWT (10)
- 定时任务 (1)
- 非技术 (1)
- Redis (5)
- Nginx (5)
- 云计算和虚拟化 (1)
- swagger (1)
- 移动端UI和框架 (5)
最新评论
-
lgh1992314：
applicationContext.xml 是 spring ...
Spring的applicationContext和spring-mvc.xml的区别
一、架构和技术介绍
1、简介
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现
2、activemq的特性
1. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
2. 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
3. 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
4. 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resourceadaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业服务器上
5. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
6. 支持通过JDBC和journal提供高速的消息持久化
7. 从设计上保证了高性能的集群,客户端-服务器,点对点
8. 支持Ajax
9. 支持与Axis的整合
10. 可以很容易得调用内嵌JMS provider,进行测试
3、下载和安装ActiveMQ
1、下载
ActiveMQ的最新版本是5.10.0，但由于我们内网下载存在问题，所以目前通过内网只能下载到5.9.0，下载地址：http://activemq.apache.org/activemq-590-release.html。
2、安装
如果是在windows系统中运行，可以直接解压apache-activemq-5.9.0-bin.zip，并运行bin目录下的activemq.bat文件，此时使用的是默认的服务端口：61616和默认的console端口：8161。
如果是在linux或unix下运行，在bin目录下执行命令：./activemq setup
3、修改ActiveMQ的服务端口和console端口
A、修改服务端口：打开conf/activemq.xml文件，修改以下红色字体部分
<transportConnectors>
<transportConnector name="openwire" uri="tcp://10.42.220.72:61618"discoveryUri="multicast://default"/>
</transportConnectors>
B、修改console的地址和端口:打开conf/jetty.xml文件，修改以下红色字体部分
<bean id="jettyPort"class="org.apache.activemq.web.WebConsolePort"init-method="start">
<property name="port" value="8162"/>
</bean>
4、通过客户端代码试用ActiveMQ
需要提前将activemq解压包中的lib目录下的相关包引入到工程中，再进行如下编码：
1、发送端的代码：
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.DeliveryMode;
importjavax.jms.Destination;
importjavax.jms.MessageProducer;
importjavax.jms.Session;
importjavax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
publicclass Sender {
privatestaticfinalintSEND_NUMBER = 5;
publicstaticvoid main(String[] args) {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection：JMS客户端到JMS Provider的连接
Connection connection = null;
// Session：一个发送或接收消息的线程
Session session;
// Destination：消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer：消息发送者
MessageProducer producer;
// TextMessage message;
//构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"failover:(tcp://10.42.220.72:61617,tcp://10.42.220.72:61618)");
try {
//构造从工厂得到连接对象
connection =connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//获取session
destination = session.createQueue("FirstQueue");
//得到消息生成者【发送者】
producer =session.createProducer(destination);
//设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//构造消息，此处写死，项目就是参数，或者方法获取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
publicstaticvoid sendMessage(Session session,MessageProducer producer)
throws Exception {
for (int i = 1; i <=SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq发送的消息" + i);
//发送消息到目的地方
System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
producer.send(message);
}
}
}
2、接收端代码：
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.Destination;
importjavax.jms.MessageConsumer;
importjavax.jms.Session;
importjavax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
publicclass Receive {
publicstaticvoid main(String[] args) {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection：JMS客户端到JMS Provider的连接
Connection connection = null;
// Session：一个发送或接收消息的线程
Session session;
// Destination：消息的目的地;消息发送给谁.
Destination destination;
//消费者，消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"failover:(tcp://10.42.220.72:61617,tcp://10.42.220.72:61618)");
try {
//构造从工厂得到连接对象
connection =connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接
session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
//获取session
destination = session.createQueue("FirstQueue");
consumer =session.createConsumer(destination);
while (true) {
//设置接收者接收消息的时间，为了便于测试，这里谁定为100s
TextMessage message =(TextMessage) consumer.receive(100000);
if (null != message) {
System.out.println("收到消息" + message.getText());
} else {
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
}
3、通过监控查看消息堆栈的记录：
登陆http://localhost:8162/admin/queues.jsp，默认的用户名和密码：admin/admin
二、ActiveMQ的多种部署方式
单点的ActiveMQ作为企业应用无法满足高可用和集群的需求，所以ActiveMQ提供了master-slave、broker cluster等多种部署方式，但通过分析多种部署方式之后我认为需要将两种部署方式相结合才能满足我们公司分布式和高可用的需求，所以后面就重点将解如何将两种部署方式相结合。
1、Master-Slave部署方式
1）shared filesystem Master-Slave部署方式
主要是通过共享存储目录来实现master和slave的热备，所有的ActiveMQ应用都在不断地获取共享目录的控制权，哪个应用抢到了控制权，它就成为master。
多个共享存储目录的应用，谁先启动，谁就可以最早取得共享目录的控制权成为master，其他的应用就只能作为slave。
2）shared database Master-Slave方式
与shared filesystem方式类似，只是共享的存储介质由文件系统改成了数据库而已。
3）Replicated LevelDB Store方式
这种主备方式是ActiveMQ5.9以后才新增的特性，使用ZooKeeper协调选择一个node作为master。被选择的master broker node开启并接受客户端连接。
其他node转入slave模式，连接master并同步他们的存储状态。slave不接受客户端连接。所有的存储操作都将被复制到连接至Master的slaves。
如果master死了，得到了最新更新的slave被允许成为master。fialed node能够重新加入到网络中并连接master进入slave mode。所有需要同步的disk的消息操作都将等待存储状态被复制到其他法定节点的操作完成才能完成。所以，如果你配置了replicas=3，那么法定大小是(3/2)+1=2. Master将会存储并更新然后等待 (2-1)=1个slave存储和更新完成，才汇报success。至于为什么是2-1，熟悉Zookeeper的应该知道，有一个node要作为观擦者存在。
单一个新的master被选中，你需要至少保障一个法定node在线以能够找到拥有最新状态的node。这个node将会成为新的master。因此，推荐运行至少3个replica nodes，以防止一个node失败了，服务中断。
2、Broker-Cluster部署方式
前面的Master-Slave的方式虽然能解决多服务热备的高可用问题，但无法解决负载均衡和分布式的问题。Broker-Cluster的部署方式就可以解决负载均衡的问题。
Broker-Cluster部署方式中，各个broker通过网络互相连接，并共享queue。当broker-A上面指定的queue-A中接收到一个message处于pending状态，而此时没有consumer连接broker-A时。如果cluster中的broker-B上面由一个consumer在消费queue-A的消息，那么broker-B会先通过内部网络获取到broker-A上面的message，并通知自己的consumer来消费。
1）static Broker-Cluster部署
在activemq.xml文件中静态指定Broker需要建立桥连接的其他Broker：
1、 首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61617)"duplex="false"/>
</networkConnectors>
2、 修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
3、 在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
4、 修改Broker-A节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
5、分别启动Broker-A和Broker-B。
2）Dynamic Broker-Cluster部署
在activemq.xml文件中不直接指定Broker需要建立桥连接的其他Broker，由activemq在启动后动态查找：
1、 首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnectoruri="multicast://default"
dynamicOnly="true"
networkTTL="3"
prefetchSize="1"
decreaseNetworkConsumerPriority="true" />
</networkConnectors>
2、修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616? " discoveryUri="multicast://default"/>
</transportConnectors>
3、在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnectoruri="multicast://default"
dynamicOnly="true"
networkTTL="3"
prefetchSize="1"
decreaseNetworkConsumerPriority="true" />
</networkConnectors>
4、修改Broker-B节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617" discoveryUri="multicast://default"/>
</transportConnectors>
5、启动Broker-A和Broker-B
2、Master-Slave与Broker-Cluster相结合的部署方式
可以看到Master-Slave的部署方式虽然解决了高可用的问题，但不支持负载均衡，Broker-Cluster解决了负载均衡，但当其中一个Broker突然宕掉的话，那么存在于该Broker上处于Pending状态的message将会丢失，无法达到高可用的目的。
由于目前ActiveMQ官网上并没有一个明确的将两种部署方式相结合的部署方案，所以我尝试者把两者结合起来部署：
1、部署的配置修改
这里以Broker-A + Broker-B建立cluster，Broker-C作为Broker-B的slave为例：
1）首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="masterslave:(tcp://0.0.0.0:61617,tcp:// 0.0.0.0:61618)" duplex="false"/>
</networkConnectors>
2）修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
3）在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
4）修改Broker-B节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
5）修改Broker-B节点中的持久化方式：
<persistenceAdapter>
<kahaDB directory="/localhost/kahadb"/>
</persistenceAdapter>
6）在Broker-C节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
7）修改Broker-C节点中的服务提供端口为61618：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61618?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
8）修改Broker-B节点中的持久化方式：
<persistenceAdapter>
<kahaDB directory="/localhost/kahadb"/>
</persistenceAdapter>
9）分别启动broker-A、broker-B、broker-C，因为是broker-B先启动，所以“/localhost/kahadb”目录被lock住，broker-C将一直处于挂起状态，当人为停掉broker-B之后，broker-C将获取目录“/localhost/kahadb”的控制权，重新与broker-A组成cluster提供服务。
1、简介
ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现
2、activemq的特性
1. 多种语言和协议编写客户端。语言: Java, C, C++, C#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
2. 完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)
3. 对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性
4. 通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resourceadaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE1.4商业服务器上
5. 支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
6. 支持通过JDBC和journal提供高速的消息持久化
7. 从设计上保证了高性能的集群,客户端-服务器,点对点
8. 支持Ajax
9. 支持与Axis的整合
10. 可以很容易得调用内嵌JMS provider,进行测试
3、下载和安装ActiveMQ
1、下载
ActiveMQ的最新版本是5.10.0，但由于我们内网下载存在问题，所以目前通过内网只能下载到5.9.0，下载地址：http://activemq.apache.org/activemq-590-release.html。
2、安装
如果是在windows系统中运行，可以直接解压apache-activemq-5.9.0-bin.zip，并运行bin目录下的activemq.bat文件，此时使用的是默认的服务端口：61616和默认的console端口：8161。
如果是在linux或unix下运行，在bin目录下执行命令：./activemq setup
3、修改ActiveMQ的服务端口和console端口
A、修改服务端口：打开conf/activemq.xml文件，修改以下红色字体部分
<transportConnectors>
<transportConnector name="openwire" uri="tcp://10.42.220.72:61618"discoveryUri="multicast://default"/>
</transportConnectors>
B、修改console的地址和端口:打开conf/jetty.xml文件，修改以下红色字体部分
<bean id="jettyPort"class="org.apache.activemq.web.WebConsolePort"init-method="start">
<property name="port" value="8162"/>
</bean>
4、通过客户端代码试用ActiveMQ
需要提前将activemq解压包中的lib目录下的相关包引入到工程中，再进行如下编码：
1、发送端的代码：
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.DeliveryMode;
importjavax.jms.Destination;
importjavax.jms.MessageProducer;
importjavax.jms.Session;
importjavax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
publicclass Sender {
privatestaticfinalintSEND_NUMBER = 5;
publicstaticvoid main(String[] args) {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection：JMS客户端到JMS Provider的连接
Connection connection = null;
// Session：一个发送或接收消息的线程
Session session;
// Destination：消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer：消息发送者
MessageProducer producer;
// TextMessage message;
//构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"failover:(tcp://10.42.220.72:61617,tcp://10.42.220.72:61618)");
try {
//构造从工厂得到连接对象
connection =connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
//获取session
destination = session.createQueue("FirstQueue");
//得到消息生成者【发送者】
producer =session.createProducer(destination);
//设置不持久化，此处学习，实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
//构造消息，此处写死，项目就是参数，或者方法获取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
publicstaticvoid sendMessage(Session session,MessageProducer producer)
throws Exception {
for (int i = 1; i <=SEND_NUMBER; i++) {
TextMessage message = session
.createTextMessage("ActiveMq发送的消息" + i);
//发送消息到目的地方
System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
producer.send(message);
}
}
}
2、接收端代码：
importjavax.jms.Connection;
importjavax.jms.ConnectionFactory;
importjavax.jms.Destination;
importjavax.jms.MessageConsumer;
importjavax.jms.Session;
importjavax.jms.TextMessage;
importorg.apache.activemq.ActiveMQConnection;
importorg.apache.activemq.ActiveMQConnectionFactory;
publicclass Receive {
publicstaticvoid main(String[] args) {
// ConnectionFactory：连接工厂，JMS用它创建连接
ConnectionFactory connectionFactory;
// Connection：JMS客户端到JMS Provider的连接
Connection connection = null;
// Session：一个发送或接收消息的线程
Session session;
// Destination：消息的目的地;消息发送给谁.
Destination destination;
//消费者，消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(
ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD,
"failover:(tcp://10.42.220.72:61617,tcp://10.42.220.72:61618)");
try {
//构造从工厂得到连接对象
connection =connectionFactory.createConnection();
//启动
connection.start();
//获取操作连接
session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
//获取session
destination = session.createQueue("FirstQueue");
consumer =session.createConsumer(destination);
while (true) {
//设置接收者接收消息的时间，为了便于测试，这里谁定为100s
TextMessage message =(TextMessage) consumer.receive(100000);
if (null != message) {
System.out.println("收到消息" + message.getText());
} else {
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection)
connection.close();
} catch (Throwable ignore) {
}
}
}
}
3、通过监控查看消息堆栈的记录：
登陆http://localhost:8162/admin/queues.jsp，默认的用户名和密码：admin/admin
二、ActiveMQ的多种部署方式
单点的ActiveMQ作为企业应用无法满足高可用和集群的需求，所以ActiveMQ提供了master-slave、broker cluster等多种部署方式，但通过分析多种部署方式之后我认为需要将两种部署方式相结合才能满足我们公司分布式和高可用的需求，所以后面就重点将解如何将两种部署方式相结合。
1、Master-Slave部署方式
1）shared filesystem Master-Slave部署方式
主要是通过共享存储目录来实现master和slave的热备，所有的ActiveMQ应用都在不断地获取共享目录的控制权，哪个应用抢到了控制权，它就成为master。
多个共享存储目录的应用，谁先启动，谁就可以最早取得共享目录的控制权成为master，其他的应用就只能作为slave。
2）shared database Master-Slave方式
与shared filesystem方式类似，只是共享的存储介质由文件系统改成了数据库而已。
3）Replicated LevelDB Store方式
这种主备方式是ActiveMQ5.9以后才新增的特性，使用ZooKeeper协调选择一个node作为master。被选择的master broker node开启并接受客户端连接。
其他node转入slave模式，连接master并同步他们的存储状态。slave不接受客户端连接。所有的存储操作都将被复制到连接至Master的slaves。
如果master死了，得到了最新更新的slave被允许成为master。fialed node能够重新加入到网络中并连接master进入slave mode。所有需要同步的disk的消息操作都将等待存储状态被复制到其他法定节点的操作完成才能完成。所以，如果你配置了replicas=3，那么法定大小是(3/2)+1=2. Master将会存储并更新然后等待 (2-1)=1个slave存储和更新完成，才汇报success。至于为什么是2-1，熟悉Zookeeper的应该知道，有一个node要作为观擦者存在。
单一个新的master被选中，你需要至少保障一个法定node在线以能够找到拥有最新状态的node。这个node将会成为新的master。因此，推荐运行至少3个replica nodes，以防止一个node失败了，服务中断。
2、Broker-Cluster部署方式
前面的Master-Slave的方式虽然能解决多服务热备的高可用问题，但无法解决负载均衡和分布式的问题。Broker-Cluster的部署方式就可以解决负载均衡的问题。
Broker-Cluster部署方式中，各个broker通过网络互相连接，并共享queue。当broker-A上面指定的queue-A中接收到一个message处于pending状态，而此时没有consumer连接broker-A时。如果cluster中的broker-B上面由一个consumer在消费queue-A的消息，那么broker-B会先通过内部网络获取到broker-A上面的message，并通知自己的consumer来消费。
1）static Broker-Cluster部署
在activemq.xml文件中静态指定Broker需要建立桥连接的其他Broker：
1、 首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61617)"duplex="false"/>
</networkConnectors>
2、 修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
3、 在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
4、 修改Broker-A节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
5、分别启动Broker-A和Broker-B。
2）Dynamic Broker-Cluster部署
在activemq.xml文件中不直接指定Broker需要建立桥连接的其他Broker，由activemq在启动后动态查找：
1、 首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnectoruri="multicast://default"
dynamicOnly="true"
networkTTL="3"
prefetchSize="1"
decreaseNetworkConsumerPriority="true" />
</networkConnectors>
2、修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616? " discoveryUri="multicast://default"/>
</transportConnectors>
3、在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnectoruri="multicast://default"
dynamicOnly="true"
networkTTL="3"
prefetchSize="1"
decreaseNetworkConsumerPriority="true" />
</networkConnectors>
4、修改Broker-B节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617" discoveryUri="multicast://default"/>
</transportConnectors>
5、启动Broker-A和Broker-B
2、Master-Slave与Broker-Cluster相结合的部署方式
可以看到Master-Slave的部署方式虽然解决了高可用的问题，但不支持负载均衡，Broker-Cluster解决了负载均衡，但当其中一个Broker突然宕掉的话，那么存在于该Broker上处于Pending状态的message将会丢失，无法达到高可用的目的。
由于目前ActiveMQ官网上并没有一个明确的将两种部署方式相结合的部署方案，所以我尝试者把两者结合起来部署：
1、部署的配置修改
这里以Broker-A + Broker-B建立cluster，Broker-C作为Broker-B的slave为例：
1）首先在Broker-A节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="masterslave:(tcp://0.0.0.0:61617,tcp:// 0.0.0.0:61618)" duplex="false"/>
</networkConnectors>
2）修改Broker-A节点中的服务提供端口为61616：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
3）在Broker-B节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
4）修改Broker-B节点中的服务提供端口为61617：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61617?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
5）修改Broker-B节点中的持久化方式：
<persistenceAdapter>
<kahaDB directory="/localhost/kahadb"/>
</persistenceAdapter>
6）在Broker-C节点中添加networkConnector节点：
<networkConnectors>
<networkConnector uri="static:(tcp:// 0.0.0.0:61616)"duplex="false"/>
</networkConnectors>
7）修改Broker-C节点中的服务提供端口为61618：
<transportConnectors>
<transportConnectorname="openwire"uri="tcp://0.0.0.0:61618?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
8）修改Broker-B节点中的持久化方式：
<persistenceAdapter>
<kahaDB directory="/localhost/kahadb"/>
</persistenceAdapter>
9）分别启动broker-A、broker-B、broker-C，因为是broker-B先启动，所以“/localhost/kahadb”目录被lock住，broker-C将一直处于挂起状态，当人为停掉broker-B之后，broker-C将获取目录“/localhost/kahadb”的控制权，重新与broker-A组成cluster提供服务。
分享到：
发表评论
-
如何用消息系统避免分布式事务？
2016-03-09 15:23 512前阵子从支付宝转账1 ... -
ActiveMQ入门实例
2015-07-13 16:28 6031.下载ActiveMQ 去官方网站下载：http://a ... -
Spring整合ActiveMQ-序列化的 Java对象
2015-04-09 16:13 1089[size=medium]JMS定义了五种不同的消息正文格式， ... -
ActiveMQ初步了解心得
2015-04-09 16:13 634使用JMS的优点: ... -
linux上安装activeMQ
2015-04-08 16:34 7911.新建一个文件夹activeMQ mkdir /serv ... -
ActiveMQ传输文件的几种方式原理与优劣
2015-04-08 15:20 4250本文讨论ActiveMQ传输文件的几种方法的原理及其利弊，作为 ... -
CentOS 5.6 安装RabbitMQ
2015-04-08 15:10 612CentOS 5.6 安装RabbitMQ步骤参考官方地址 ... -
消息中间件的技术选型心得－RabbitMQ、ActiveMQ和ZeroMQ
2015-04-08 15:08 870作者：chszs，转载需注明。博客主页：http://blo ... -
RabbitMQ的安装，配置，监控 (转)
2015-04-08 14:51 561安装 RabbitMQ是基于Erlang的，所以首先必须配 ...
相关推荐
本篇文章将详细讲解如何搭建ActiveMQ的高可用和负载均衡集群，以及进行相关的安装配置和高可用测试。 一、ActiveMQ简介 ActiveMQ是一款功能强大的消息代理，它支持多种消息协议，如OpenWire、AMQP、STOMP、XMPP等，...
在高可用性和高并发场景下，为了实现负载均衡和故障转移，ActiveMQ 提供了 Broker-Cluster（经纪人集群）部署模式。这种部署方式允许多个 ActiveMQ 经纪人实例通过网络互相连接，共享队列，从而实现消息的负载均衡。...
在本文中，我们将深入探讨ActiveMQ的高可用性和负载均衡集群的实现，以及如何通过源码和文档来理解这些概念。 首先，让我们了解一下ActiveMQ的高可用性。高可用性（High Availability，简称HA）是指系统能够在硬件...
三、ActiveMQ高可用部署 ActiveMQ作为Java消息服务（JMS）的实现，用于实现应用程序之间的异步通信。在Linux下部署ActiveMQ集群，以提高消息传递的可靠性和性能： 1. 安装ActiveMQ：下载并解压ActiveMQ至指定目录...
本主题将深入探讨如何使用Apache ActiveMQ，一个流行的开源消息代理，来实现高可用性（HA）和负载均衡集群。以下是对相关知识点的详细说明： 1. **Apache ActiveMQ**：ActiveMQ是基于Java的消息中间件，遵循JMS...
本文将深入探讨如何安装、配置ActiveMQ以实现高可用性集群，并进行负载均衡测试，同时利用ZooKeeper和LevelDB来增强系统的稳定性和可靠性。 首先，ActiveMQ的高可用性通常通过集群实现，集群中的每个节点都是...
无论是通过连接池管理资源、调整预取策略来实现消息的均衡分配，还是通过垂直扩展和水平扩展来提高系统的整体吞吐量，都是确保ActiveMQ能够在高负载环境下稳定运行的关键。此外，对于特定的需求，还可以根据实际情况...
在Windows Server环境下构建Zookeeper和ActiveMQ的集群部署是一个实现高可用性和负载均衡的重要实践。以下将详细阐述这个过程中的关键知识点。 首先，我们需要安装Java Development Kit (JDK)。确保下载并安装与...
ActiveMQ支持高可用、集群、故障转移和持久化消息等特性，可以用于负载均衡和提高系统可靠性。它的设计目标是为了解决分布式系统中的消息传递和集成问题，同时也支持通过多种协议进行消息通信，如TCP/IP、SSL、NIO、...
总结来说，ActiveMQ的集群技术提供了高可用性和负载均衡的能力，通过Master-Slave和Broker Cluster两种模式实现了消息服务的稳定运行。在客户端方面，failover://协议允许客户端在Broker失效时自动进行故障转移。而...
《ActiveMQ集群》和《ActiveMQ高可用+负载均衡集群的安装、配置、高可用测试》可能涉及： 1. 集群配置：ActiveMQ集群可以通过网络连接多台服务器，共享队列和主题，提供高可用性和负载均衡。 2. 高可用性：通过...
3. **高可用性**：通过集群和复制策略，ActiveMQ可以实现故障转移和负载均衡，保证服务的连续性。 4. **消息优先级与时间戳**：消息可以根据优先级进行处理，同时，系统会根据时间戳自动清理过期消息，以保持高效...
分布式架构是现代大型互联网企业中不可或缺的技术，它通过将应用程序的不同部分部署在不同的物理或虚拟节点上，以实现系统的可扩展性和高可用性。在本架构中，`Dubbo`是一个关键的分布式服务治理框架，它允许服务...
在实际应用中，可能还需要考虑消息的持久化、负载均衡、高可用性等问题，以便在系统故障时保证数据不丢失。 在Spring应用中整合ActiveMQ，不仅可以实现异步处理，提高系统的响应速度，还能实现解耦，使得各个组件...
此外，还深入解析了ActiveMQ的核心特性，如网络连接、负载均衡、消息路由、高可用性集群和安全性设置。对于开发者来说，理解这些概念和功能对于构建稳定、高效的消息传递系统至关重要。 源码部分则提供了ActiveMQ的...
4. **集群和高可用性**：通过组建集群，ActiveMQ可以实现负载均衡和故障转移，提高服务的可用性。 5. **安全控制**：ActiveMQ支持多种安全机制，包括SSL/TLS加密、用户认证和授权，确保消息传输的安全。 6. **管理...
3. 负载均衡与高可用：通过集群部署，ActiveMQ可以实现负载均衡和故障转移，确保服务的高可用性。 4. 安全性：支持用户认证和授权，可以设置访问控制策略，保障消息传输的安全。 5. 监控与管理：提供Web管理控制台，...
3. **高可用性**：通过集群和复制策略，ActiveMQ可以构建高可用的消息系统，确保服务的连续性。 4. **网络传输**：ActiveMQ支持跨网络的部署和消息路由，使得分布式系统能够有效地通信。 5. **管理工具**：...
3. **高可用性**：通过集群和复制策略，ActiveMQ 可以实现多个节点间的负载均衡和故障切换，确保服务的连续性。 4. **高并发处理**：通过消息的异步处理，ActiveMQ 能够处理大量的并发请求，减轻服务器压力，提高...
ActiveMQ 作为一种高性能的消息中间件，通过构建集群来实现高可用性和负载均衡是常见实践之一。本文档将详细介绍如何利用 ZooKeeper 实现 ActiveMQ 的 Master-Slave 集群配置，以及具体的部署步骤。 #### 二、...