大家好，欢迎来到IT知识分享网。

Activemq的集群讲解

1. 了解ActiveMQ

ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在早些年的 “J2EE应用” 时期扮演着特殊的地位，可以说那个年代ActiveMQ在业界应用最广泛，当然如果现在想要有更强大的性能和海量数据处理能力，ActiveMQ还需要不断的升级版本，不断的提升性能和架构设计的重构。

1.1什么是消息队列？

也就是MQ(Message Queue)，是基础数据结构中“先进先出”的一种数据结构。一般用来解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构

2．为什么需要消息队列

主要原因是由于在高并发环境下，同步请求来不及处理，请求往往会发生阻塞。大量的请求到达访问数据库，导致行锁表锁，最后请求线程会堆积过多，从而触发 too many connection错误，引发雪崩效应。我们使用消息队列，通过异步处理请求，从而缓解系统的压力。核心：异步处理、流量削峰、应用解耦

2.1应用场景

异步处理，流量削峰，应用解耦，消息通讯四个场景

2.1.1异步处理

• 场景1：用户注册后，需要发送注册邮件和注册短信。
  • 串行方式：将注册信息写入 数据库 成功后，发送注册邮件，再发送注册短信。以上三个任务全部完成后，返回给客户端

• 并行方式：将注册信息写入数据库成功后，发送注册邮件的同时，发送注册短信。以上三个任务完成后，返回给客户端。与串行的差别是，并行的方式可以提高处理的时间
• 

假设三个业务节点每个使用50毫秒钟，不考虑网络等其他开销，则串行方式的时间是150毫秒，并行的时间可能是100毫秒。

IT知识分享网

IT知识分享网因为CPU在单位时间内处理的请求数是一定的，假设CPU在1秒内吞吐量是100次。则串行方式1秒内CPU可处理的请求量是7次（1000/150）。

并行方式处理的请求量是10次（1000/100）

• 小结：如以上案例描述，传统的方式系统的性能（并发量，吞吐量，响应时间）会有瓶颈。如何解决这个问题？
  • 引入消息队列，将不是必须的业务逻辑，异步处理。改造后的架构如下：

按照以上约定，用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回， 因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS比串行提高了3倍，比并行提高了2倍。

• 场景2：订单处理，前端应用将订单信息放到队列，后端应用从队列里依次获得消息处理，高峰时的大量订单可以积压在队列里慢慢处理掉。

2.1.2流量削峰

流量削峰也是消息队列中的常用场景，一般在 秒杀或团抢活动 中使用广泛

• 应用场景：秒杀活动，一般会因为流量过大，导致流量报增，应用挂掉。为解决这个问题，一般需要在应用前端加入消息队列。
  • 可以控制活动的人数
  • 可以缓解短时间内高流量压垮应用
  • 

• 让消息不直接到达服务器，而是先让 「消息队列」 保存这些数据，然后让下面的服务器每一次都取各自能处理的请求数再去处理，这样当请求数超过服务器最大负载时，就不至于把服务器搞挂了。
• 秒杀业务根据消息队列中的请求信息，再做后续处理

2.1.3应用解耦

• 场景说明：用户下单后，订单系统需要通知库存系统。传统的做法是，订单系统调用系统库存接口。
• 

• 传统模式的缺点：
  • 假如库存系统无法访问，则订单减库存将失败，从而导致订单失败
  • 解决订单系统与库存系统耦合，如何解决？

引入消息队列后的方案：

IT知识分享网订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功

库存系统：订阅下单的消息，采用pub/sub(发布/订阅)的方式，获取下单信息，库存系统根据下单信息，进行库存操作

• 假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入信息队列就不再关心其他后续操作了。实现订单系统与库存系统的应用解耦。

2.1.4日志处理

日志处理是将消息队列用在日志处理中，比如Kafka的应用，解决大量日志传输的问题。架构简化如下

• 日志采集客户端，负载日志数据采集，定时写入Kafka队列
• Kafka消息队列，负载日志数据的接收，储存和转发
• 日志处理应用：订阅并消费Kafka队列中的日志数据

• (1) Kafka：接收用户日志的消息队列
• (2) Logstash：做日志解析，统一成JSON 输出给Elasticsearch
• (3)Kibana：基于Elasticsearch 的数据可视化组件，超强的数据可视化能力是众多公司选择Elkstack的重要原因

2.1.5消息通讯

消息通讯是指，消息队列一般都内置了高效的通信机制就，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等。

• 点对点通讯：
• 

客户端A和客户端B使用同一队列，进行消息通讯。

• 聊天室通讯：

客户端A，客户端B，客户端N订阅同一主题，进行消息发布和接收。实现类似聊天室效果。以上实际是消息队列的两种消息模式，点对点或发布订阅模式

3.消息投递模式

我们首先要了解JMS规范里最经典的两种消息投递模式，即 “点对点” 与 “发布订阅”。

点对点：生产者向队列投递一条消息，只有一个消费者能够监听得到这条消息（PTP)

发布订阅：生产者向队列投递一条消息，所有监听该队列的消费者都能够监听得到这条消息（P/S)，下图所示：

4.activeMQ的各项指标

衡量一个MOM，我们主要从三方面考虑即可，即服务性能、存储堆积能力、可扩展性。

• 服务性能
  • ActiveMQ的性能一般，在早期传统行业为王的时代还是比较流行的，但现如今面对高并发、大数据的业务场景，往往力不从心！
• 数据存储
  • 默认采用kahadb存储（索引文件形式存储），也可以使用高性能的google leveldb（内存数据库存储）， 或者可以使用MySql、Oracle进程消息存储（关系型数据库存储）。
• 集群架构
  • ActiveMQ 可以与zookeeper进行构建 主备集群模型，并且多套的主备模型直接可以采用Network的方式构建分布式集群。

5.ActiveMQ集群架构模式

ActiveMQ最经典的两种集群架构模式，Master-Slave 、Network 集群模式

Master-Slave：

就是主从方式，当然这里要理解为主备的方式，也就是双机热备机制；Master Slave 背后的想法是，消息被复制到slave broker，因此即使master broker遇到了像硬件故障之类的错误，你也可以立即切换到slave broker而不丢失任何消息。 Master Slave是目前ActiveMQ推荐的高可靠性和容错的解决方案

架构思考：

• 上图（Master-Slave）绿色的为主节点，灰色的则为备份节点，这两个节点都是运行状态的。
• zookeeper的作用就是为了当绿色的主节点宕机时，进行及时切换到备份的灰色节点上去，使其进行主从角色的互换，用于实现高可用性的方案。
• Master-Slave集群模型的缺点也显而易见，就是不能做到分布式的topic、queue，当消息量巨大时，我们的MQ集群压力过大，没办法满足分布式的需求。

Network：

这里可以理解为网络通信方式，也可以说叫Network of brokers。这种方式真正解决了分布式消息存储和故障转移、broker切换的问题。可以理解消息会进行均衡；从ActiveMQ1.1版本起，ActiveMQ支持networks of brokers。它支持分布式的queues和topics。一个broker会相同对待所有的订阅（subscription）：不管他们是来自本地的客户连接，还是来自远程broker，它都会递送有关的消息拷贝到每个订阅。远程broker得到这个消息拷贝后，会依次把它递送到其内部的本地连接上。

Network集群模型的关键点：

首先，这种方案需要两套或多套（Master-Slave）的集群模型才可以搞定，部署非常麻烦，需要两套或多套集群直接相互交叉配置，相互间能够感知到彼此的存在。下面我给出一段XML配置，简单来说就是在ActiveMQ的配置文件里要进行多套（Master-Slave）之间的 networkConnector配置工作：

<broker brokerName=”receiver” persistent=”false” useJmx=”false”>

    <transportConnectors>

       <transportConnector uri=”tcp://localhost:62002″/>

    </transportConnectors>

    <networkConnectors>

       <networkConnector

              uri=”static:( tcp://localhost:61616,tcp://remotehost:61616)”/>

    </networkConnectors>

</broker>

其次，Network虽然解决了分布式消息队列这个难题，但是还有很多潜在的问题，最典型的就是资源浪费问题，并且也可能达不到所预期的效果；通常采用Master-Slave模型是传统型互联网公司的首选，作为互联网公司往往会选择开箱即用的消息中间件，从运维、部署、使用各个方面都要优于ActiveMQ，当然ActiveMQ毕竟是 “老牌传统强Q”，Apache的顶级项目之一，目前正在进行新版本的重构（对于5.X版本）与落地，下一代 “Artemis代理”，也可以理解为 “6.X”；

5.activeMQ修改控制台端口，和消息端口

5.1 jetty.xml

目录： conf/jetty.xml 修改的是控制台端口

5.2 activemq.xml

目录：conf/activemq.xml修改的是消息通信端口

6.activeMQ集群的几种方式

6.1单机伪集群(热备部署)

集群搭建在同一台虚拟机上，3个Activemq分别使用不同的端口提供服务，启用1个为Master，其它2个为Slaver，同一时间仅Master队列提供服务

3个Activemq服务,同一时间仅Master队列提供服务，当Master队列挂掉后，其它2个Slaver自动选举出1个成为Master，整个队列服务依然可用。当挂掉的队列重新恢复后，自动加入集群。当集群仅剩下1个队列时，整个队列不可用。

Activemq集群数据存储方式

      a) kahaDB：文件共享，默认方式（共享文件系统例如SAN）

      b) JDBC：数据库共享，支持MySql、Sql Server、Oracle等

      c) LevelDB：数据共享，比kahaDB更快，基于索引

基于 ZooKeeper + LevelDB 的 Master-Slave方式 

伪集群服务器配置

Zookeeper单机节点：121.36.65.132:2181

1.修改jetty.xml 修改端口

2.修改3个节点的集群名称，都为 “my-activemq-cluster”

3个activemq节点的brokerName必须一致，否则不能加入到集群中

3.修改activemq.xml persistenceAdapter节点

改为：

<persistenceAdapter>

    <replicatedLevelDB

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

          replicas=”3″

          bind=”tcp://0.0.0.0:62626″

          zkAddress=”121.36.65.132:2181″

          zkSessionTimeout=”4s”

          hostname=”121.36.65.132″

          sync=”local_disk”

          zkPath=”/activemq/leveldb-stores”

    />

</persistenceAdapter>

上述配置的意义解释如下：

directory：表示ActiveMQ集群消息持久化保存到服务器上的路径，注意该路径一定要先创建好。

replicas：表示ActiveMQ集群的节点个数。

bind：表示当这个节点成为master后，绑定的机器的地址与端口。此处0.0.0.0:0表示绑定到本机所有可用IP，而端口是随机的。

zkAddress：表示ZooKeeper的ip和port。如果是ZooKeeper集群的话，则用逗号隔开。

zkSessionTimeout：表示ActiveMQ与ZooKeeper集群连接的会话超时时间。

hostname：表示本机的IP地址。服务器根据不同的IP地址做出改变，其他配置相同。

sync：在消息被消费完成前，同步信息所存贮的策略。如果有多种策略用逗号隔开，ActiveMQ会选择较强的策略。而如果有local_mem, local_disk这两种策略的话，那么ActiveMQ则优先选择local_disk策略，存储在本地硬盘。

zkPath：表示ActiveMQ在ZooKeeper集群上创建的znode节点的路径，也即是ZooKeeper选举信息交换的存贮路径。

4. ActiveMQ集群启动完毕，根据ZooKeeper的策略，会从这三台ActiveMQ服务器选一台作为master对外提供服务，其他两台作为slave等待运行，而slave只是做数据上的主从同步。

所以，ActiveMQ集群后，访问

http://121.36.65.132:8166/admin/index.jsp

http://121.36.65.132:8167/admin/index.jsp

http://121.36.65.132:8168/admin/index.jsp

只会有一个成功。

5.那么现在在ActiveMQ集群中，如何查看哪一台服务器是master节点呢？

单机伪集群

netstat -tunlp |grep 8166

netstat -tunlp |grep 8167

netstat -tunlp |grep 8168

6.如何验证集群的高可用，kill 一个 activemq,再去访问

6.2高可用集群（多机，分流）

集群A

集群B

Zookeeper节点：119.45.63.161:2181,119.45.63.161:2182,119.45.63.161:2183

6.2.1集群A修改

activemq_1，activemq_2，activemq_3分别修改jetty.xml

分别改为8163，8164，8165

activemq_1，activemq_2，activemq_3分别修改activemq.xml

broker节点

persistenceAdapter节点

transportConnectors节点

6.2.2集群B修改

修改jetty.xml

修改activemq.xml

6.2.3zookeeper集群

上面是单机伪群，必须为奇数个

需要注意修改

myid

每个服务器 在对应的zookeeper目录下的data ,新建myid, 里面写server的id  如1,2,3

6.2.4集群之间的通讯

这里以A.B集群为例

6.2.3.1networkConnectors（网络连接模式）

服务器S1和S2通过NewworkConnector相连，则生产者P1发送消息，消费者C3和C4都可以接收到，而生产者P3发送的消息，消费者C1和C2同样也可以接收到，要使用NewworkConnector的功能，需要在服务器S1的activemq.xml中的broker节点下添加如下配置（注：10.79.11.172:61617为S2的地址）：

<networkConnectors>

    <networkConnector uri="static:(tcp://219.140.171.171:51517,tcp://219.140.171.171:51518)"/>

</networkConnectors>

6.3zoolnspector工具的使用