ActiveMQ 集群搭建与高可用方案设计（一）

一、引言

在当今分布式系统盛行的时代，消息中间件扮演着至关重要的角色，而 ActiveMQ 作为一款开源的、功能强大的消息中间件，在众多项目中得到了广泛应用。它支持多种消息传输协议，如 JMS、AMQP、MQTT 等 ，能够方便地实现系统之间的异步通信，有效解耦不同的业务模块，提升系统的可靠性和灵活性。

随着业务的不断发展和用户量的持续增长，对消息中间件的性能、可靠性和可用性提出了更高的要求。单节点的 ActiveMQ 在面对高并发、大规模数据传输以及硬件故障等情况时，往往显得力不从心。例如，在电商促销活动期间，大量的订单消息、库存变更消息等需要及时处理，如果单节点 ActiveMQ 出现故障，可能会导致订单丢失、库存数据不一致等严重问题，给企业带来巨大的经济损失。

为了应对这些挑战，搭建 ActiveMQ 集群并设计高可用方案成为了必然选择。通过集群部署，可以实现负载均衡，将大量的消息请求均匀地分配到各个节点上，避免单个节点因负载过高而出现性能瓶颈。同时，集群中的多个节点可以相互备份，当某个节点发生故障时，其他节点能够迅速接管其工作，保证消息的可靠传输和系统的持续运行，从而大大提高了系统的可用性和容错能力 。

在接下来的内容中，我们将深入探讨 ActiveMQ 集群搭建的详细步骤以及高可用方案的设计与实现，希望能为大家在实际项目中应用 ActiveMQ 提供有价值的参考。

二、ActiveMQ 基础入门

（一）ActiveMQ 简介

ActiveMQ 是 Apache 软件基金会下的一款开源的消息中间件，基于 Java 语言开发 ，能够在分布式系统中高效地实现消息的发送和接收。它支持多种消息传输协议，像 JMS（Java Message Service）、AMQP（Advanced Message Queuing Protocol）、MQTT（Message Queuing Telemetry Transport） 等，这使得它可以在不同的技术架构和应用场景中灵活应用。

ActiveMQ 具备多种特性，为企业级应用提供了可靠的消息通信支持。它支持多种语言编写客户端，包括 Java、C、C++、C#、Ruby、Perl、Python、PHP 等，方便不同技术栈的开发团队进行集成。同时，ActiveMQ 完全支持 JMS1.1 和 J2EE 1.4 规范，涵盖了持久化、XA 消息、事务等重要功能，确保了消息的可靠传输和事务一致性 。在消息持久化方面，ActiveMQ 支持通过 JDBC 和 journal 提供高速的消息持久化，保证消息在系统故障等情况下不丢失。并且，它从设计上保证了高性能的集群、客户端 - 服务器、点对点的通信模式，能够满足不同业务场景下的通信需求。

在应用场景方面，ActiveMQ 有着广泛的应用。在电商系统中，当用户下单后，订单系统可以将订单消息发送到 ActiveMQ 队列，库存系统、物流系统等作为消费者从队列中获取消息并进行相应处理，如库存扣减、物流单生成等，实现了不同业务系统之间的解耦和异步通信。在分布式系统的日志处理中，各个服务产生的日志信息可以通过 ActiveMQ 发送到日志收集系统，由日志收集系统统一进行存储和分析，提高了日志处理的效率和灵活性 。在消息推送场景中，如新闻资讯推送、即时通讯消息推送等，ActiveMQ 可以作为消息的中转站，将消息推送给多个订阅者，实现一对多的消息广播。

（二）核心概念

1. 消息（Message）：消息是 ActiveMQ 中最基本的概念，它是一种包含数据和元数据的对象。消息的元数据包含消息 ID，用于唯一标识消息；消息属性，比如优先级、时间戳等，优先级高的消息可能会被优先处理，时间戳可以记录消息的发送或接收时间；消息体则是消息的具体内容，可以是文本消息，如 XML、JSON 格式的数据，也可以是二进制消息，像图片、音频等。例如，在一个订单消息中，消息体可能包含订单编号、商品信息、下单用户等具体内容，而消息属性中可以设置该订单消息的优先级，以便在处理时优先处理重要订单 。

1. 队列（Queue）：队列是 ActiveMQ 中的一个消息代理，用于存储和管理消息。队列中的消息按照先进先出（FIFO）的原则进行排序，就像我们日常生活中排队买票一样，先到的人先办理业务。生产者将消息发送到队列，消费者从队列中取消息进行处理。一个典型的场景是订单处理，订单系统作为生产者将订单消息发送到队列，而订单处理系统作为消费者从队列中依次获取订单消息进行处理，确保每个订单都能被有序地处理，并且每个消息只会被一个消费者消费 。

1. 主题（Topic）：主题与队列类似，也是一个消息代理。不同之处在于，主题采用发布 / 订阅模式。一个生产者可以向主题发送消息，多个消费者可以订阅该主题，并且每个订阅者都会接收到相同的消息，就如同广播一样，所有收听广播的人都能听到相同的内容。例如，在新闻发布系统中，新闻发布者将新闻消息发布到主题，各个新闻客户端作为消费者订阅该主题，这样所有订阅的新闻客户端都能收到最新的新闻消息 。

1. 生产者（Producer）：生产者是负责将消息发送到队列或主题的组件，它可以是一个进程、线程或者是一个应用程序。在电商系统中，订单系统就可以看作是生产者，当用户下单时，订单系统会生成订单消息并将其发送到 ActiveMQ 的队列或主题中，以便其他系统进行后续处理 。

1. 消费者（Consumer）：消费者是从队列或主题中取消息进行处理的组件，同样可以是一个进程、线程或者应用程序。比如在上述电商系统中，库存系统和物流系统就是消费者，它们分别从队列或主题中获取订单消息，库存系统根据消息进行库存扣减操作，物流系统根据消息生成物流单并安排发货 。

三、集群搭建前的准备工作

（一）环境准备

1. 硬件环境：为了确保 ActiveMQ 集群能够稳定运行并具备良好的性能，建议选择配置较高的服务器。每台服务器至少配备 2 核及以上的 CPU，这能够保证在处理大量消息请求时，服务器有足够的计算资源来高效地进行消息的接收、存储和转发。内存方面，4GB 及以上是较为合适的，充足的内存可以避免因内存不足导致的性能瓶颈，确保消息在内存中能够快速地被处理和传输。硬盘方面，推荐使用高速的固态硬盘（SSD），其读写速度远远高于传统的机械硬盘，能够显著提升消息持久化和读取的速度，保证消息的可靠存储和快速获取。网络方面，服务器之间应通过千兆及以上的网络连接，以确保消息在集群节点之间能够快速、稳定地传输，减少网络延迟对消息处理的影响 。

1. 软件环境：操作系统可以选择 Linux 系统，如 CentOS 7 或更高版本，Linux 系统具有良好的稳定性、安全性和开源特性，能够为 ActiveMQ 集群提供可靠的运行环境。同时，需要安装 Java 环境，因为 ActiveMQ 是基于 Java 开发的，Java 环境是其运行的基础。建议安装 Java Development Kit（JDK）1.8 及以上版本，安装完成后，需要配置好 JAVA_HOME、PATH 和 CLASSPATH 等环境变量，确保系统能够正确识别和使用 Java。例如，在 CentOS 系统中，可以通过编辑 /etc/profile 文件，添加如下配置：

export JAVA_HOME=/usr/local/jdk1.8.0_311 # 根据实际安装路径修改

export PATH=$JAVA_HOME/bin:$PATH

export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar

配置完成后，执行source /etc/profile命令使配置生效 。

（二）下载与安装 ActiveMQ

1. 下载途径：可以从 ActiveMQ 的官方网站（http://activemq.apache.org/components/classic/download/ ）下载最新版本的 ActiveMQ 安装包。在下载页面，根据操作系统的类型选择对应的安装包，如 Windows 系统选择.zip格式的安装包，Linux 系统选择.tar.gz格式的安装包 。

1. 安装步骤：

• • Windows 系统：下载完成后，解压安装包到指定目录，例如D:\activemq。解压完成后，进入bin\win64目录（如果是 32 位系统，则进入bin\win32目录），该目录下有多个脚本文件，其中activemq.bat是启动 ActiveMQ 的脚本，wrapper.exe也可用于启动服务，InstallService.bat用于将 ActiveMQ 安装为系统服务，UninstallService.bat用于卸载系统服务。可以根据需求选择启动方式，若选择将其安装为系统服务，双击InstallService.bat即可，安装完成后，在系统服务列表中可以找到 ActiveMQ 服务，可通过服务管理器进行启动、停止和管理 。

• • Linux 系统：将下载的.tar.gz安装包上传到服务器，例如/opt目录。使用命令tar -zxvf apache-activemq-5.16.3-bin.tar.gz（假设下载的版本是 5.16.3，根据实际版本修改）解压安装包，解压后会生成一个apache-activemq-5.16.3目录。进入解压后的目录，再进入bin/linux-x86-64目录（根据服务器架构选择对应目录），执行./activemq start命令启动 ActiveMQ。启动过程中，可能会因为防火墙的限制导致无法访问管理页面和消息端口，需要开放相应的端口。例如，使用firewall - cmd命令开放 8161（管理页面端口）和 61616（消息端口）：

# 开放8161端口策略

firewall - cmd --zone = public --add - port = 8161/tcp --permanent

# 开放61616端口策略

firewall - cmd --zone = public --add - port = 61616/tcp --permanent

# 重新载入防火墙策略

firewall - cmd --reload

启动成功后，可以通过http://服务器IP:8161/admin访问 ActiveMQ 的管理页面，默认用户名和密码都是admin 。

（三）配置文件说明

ActiveMQ 的主要配置文件是conf/activemq.xml，该文件包含了 ActiveMQ 的各种核心配置参数 。

1. broker 元素：broker元素是activemq.xml文件的核心元素之一，它定义了 ActiveMQ 代理的基本属性。其中，brokerName属性用于指定代理的名称，在集群环境中，所有节点的brokerName必须相同，这样才能确保它们属于同一个集群，例如brokerName="myActiveMQCluster" 。dataDirectory属性指定了 ActiveMQ 的数据存储目录，默认值是${activemq.data}，该目录用于存储消息持久化数据、日志文件等重要信息 。

1. destinationPolicy 元素：destinationPolicy元素用于配置目的地（队列或主题）的相关策略。在destinationPolicy元素内部，可以通过policyMap和policyEntries元素来定义针对不同目的地的具体策略。例如，可以设置消息的存储限制、转发策略等。对于主题（topic），可以设置producerFlowControl属性为true，表示启用生产者流控制，当主题中的消息堆积过多时，生产者将被阻塞，防止消息的无限堆积；optimizedDispatch属性设置为true，可以优化消息的分发策略，提高消息的处理效率 。

1. persistenceAdapter 元素：persistenceAdapter元素用于配置消息的持久化方式。ActiveMQ 支持多种持久化方式，如kahadb、leveldb等。默认情况下，使用kahadb进行持久化，配置如下：

<persistenceAdapter>

<kahaDB directory="${activemq.data}/kahadb"/>

</persistenceAdapter>

在集群环境中，通常会使用基于 Zookeeper 和 LevelDB 的持久化方式，这种方式能够提供更好的高可用性和数据一致性。配置示例如下：

<persistenceAdapter>

<replicatedLevelDB

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

replicas="3"

bind="tcp://0.0.0.0:62222"

zkAddress="192.168.1.100:2181,192.168.1.101:2181,192.168.1.102:2181"

hostname="192.168.1.100"

zkPath="/activemq/leveldb-stores/"/>

</persistenceAdapter>

其中，directory指定了 LevelDB 数据存储目录；replicas表示集群中的节点数量，一般设置为奇数，以确保在节点故障时能够正常选举出主节点；bind指定了集群内部通信的端口；zkAddress指定了 Zookeeper 集群的地址；hostname指定了当前节点的 IP 地址；zkPath指定了 ActiveMQ 在 Zookeeper 中存储选举信息和数据同步的路径 。

4. transportConnectors 元素：transportConnectors元素用于配置 ActiveMQ 的传输连接器，定义了客户端与 ActiveMQ 代理进行通信的协议和端口。例如，配置 TCP 协议的传输连接器：

<transportConnectors>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>

</transportConnectors>

其中，name指定了传输连接器的名称；uri指定了通信的协议（这里是 TCP）、地址（0.0.0.0表示监听所有 IP 地址）和端口（61616），还可以设置一些参数，如maximumConnections表示最大连接数，wireFormat.maxFrameSize表示最大帧大小 。除了 TCP 协议，ActiveMQ 还支持 AMQP、STOMP、MQTT 等多种协议，可根据实际需求进行配置 。

四、ActiveMQ 集群搭建实战

（一）Master - Slave 模式搭建

1. Pure Master Slave：在 Pure Master Slave 模式中，整个集群由一个 Master 节点和一个 Slave 节点组成。运行时，Slave 节点通过网络实时从 Master 节点复制数据，以此保证消息的一致性和可靠性 。当 Slave 和 Master 失去连接时，Slave 会自动升级为 Master，继续为客户端提供消息服务。

这种模式的优点是实现相对简单，能够提供基本的高可用性保障，适合对集群复杂性要求不高、规模较小的应用场景，在一些小型企业的内部系统中，如果对消息服务的可靠性有一定要求，但又不想投入过多的资源进行复杂的集群配置，就可以采用这种模式 。然而，它也存在明显的缺点，由于只有一个 Master 和一个 Slave，扩展能力非常有限，难以满足大规模业务增长的需求；并且在 Master 失效后，若想恢复 Master，需要停止 Slave，拷贝 Slave 中的数据文件到 Master 中，然后重启，操作过程较为繁琐 。

在配置方面，Master 不需要做特殊配置。对于 Slave broker，在${ACTIVEMQ_HOME}/conf/activemq.xml文件的<broker>节点中添加连接到 Master 的 URI 和设置 Master 失效后不关闭 Slave，示例如下：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="pure_slave"

masterConnectorURI="tcp://0.0.0.0:61616" shutdownOnMasterFailure="false" dataDirectory="${activemq.base}">

同时，需要修改 Slave 的服务端口，以避免与 Master 端口冲突，示例如下：

<transportConnectors>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61617"/>

</transportConnectors>

不过需要注意的是，从 ActiveMQ 5.8 版本开始，Pure Master Slave 模式已被废弃，不再推荐使用 。

2. Shared File System Master Slave：基于共享文件系统实现的 Master - Slave 模式，其原理是利用共享文件系统来存储消息数据。ActiveMQ 的默认数据库 kahaDB 底层是文件系统，多个 ActiveMQ 实例共享同一个文件存储目录。哪个 ActiveMQ 实例先获取共享文件的锁，那个实例就是 Master，其它的 ActiveMQ 实例就是 Slave。当当前的 Master 失效时，其它的 Slave 就会去竞争共享文件锁，谁竞争到了谁就是新的 Master 。

这种模式的优势在于 Slave 的个数没有限制，具有较好的扩展性；并且当 Master 失效时，无需手动配置，只要有足够多的 Slave，就可以自动选举出新的 Master，提高了系统的可用性 。在配置共享文件系统时，如果是在同一台机器上进行测试，可以将持久化适配器的存储目录改为本地磁盘的一个固定目录，多个实例共享这个目录，示例如下：

<persistenceAdapter>

<kahaDB directory="E:/XXX/XXX/XXX/cluster/shared_file/data/kahadb"/>

</persistenceAdapter>

如果各个 ActiveMQ 实例需要运行在不同的机器上，就需要使用分布式文件系统，如 NFS（Network File System）、Ceph 等。在 broker 配置方面，除了上述持久化适配器的配置外，还需要修改每个 broker 实例的服务端口和 jetty 的服务端口，防止端口占用异常 。

3. JDBC Master Slave：该模式使用数据库作为持久化存储来实现主从模式。其原理与 Shared File System Master Slave 模式类似，多个 ActiveMQ 实例连接到同一个共享数据库，通过争抢数据库连接并锁定数据库操作权限来确定 Master 和 Slave。争抢到连接的称为 Master，其他 broker 称为 Slave 。如果当前 Master 挂掉，另外几个 Slave 会继续争抢，挂掉的 broker 重连成功后仍然作为 Slave 。

在配置时，首先需要在所有的 ActiveMQ 的主配置文件（${ACTIVEMQ_HOME}/conf/activemq.xml）中添加数据源，所有的数据源都指向同一个数据库。以 MySQL 为例，配置示例如下：

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">

<property name="driverClassName" value="com.mysql.jdbc.Driver"/>

<property name="url" value="jdbc:mysql://localhost:3306/cluster_jdbc?relaxAutoCommit=true"/>

<property name="username" value="root"/>

<property name="password" value="root"/>

<property name="maxActive" value="200"/>

<property name="poolPreparedStatements" value="true"/>

</bean>

然后修改持久化适配器，使用jdbcPersistenceAdapter，示例如下：

<persistenceAdapter>

<jdbcPersistenceAdapter dataSource="#mysql-ds"/>

</persistenceAdapter>

这种方式的集群相对 Shared File System Master Slave 更加简单，更容易进行分布式部署。但是，如果数据库失效，那么所有的 ActiveMQ 实例都将失效，因此对数据库的可靠性要求较高 。在实际应用中，通常会结合数据库的主从复制、集群等技术来提高数据库的可靠性 。

（二）Broker - Cluster 模式搭建

1. 静态 Broker - Cluster：静态 Broker - Cluster 采用静态配置的方式，通过硬编码所有已知 ActiveMQ 实例节点的 URI 地址，在多个 ActiveMQ 实例之间进行消息的路由 。其原理是，在一个 broker 的实例内配置networkConnector，指向其他 broker 的 URI，这样各个 broker 之间就可以建立连接并共享队列和消费者列表，实现消息的转发和共享 。

在activemq.xml中配置static network Connector的示例如下：

<networkConnectors>

<networkConnector name="local network" uri="static://(tcp://remotehost1:61616,tcp://remotehost2:61616)"/>

</networkConnectors>

在这个示例中，name属性为该网络连接器指定了一个名称，方便管理和识别；uri属性指定了要连接的远程 broker 的地址，这里通过static协议指定了两个远程 broker 的 TCP 地址和端口 。当一个 broker 接收到消息后，如果本地没有对应的消费者，它会根据配置的networkConnector将消息转发到其他 broker 上，由其他 broker 上的消费者进行处理 。

2. 动态 Broker - Cluster：动态 Broker - Cluster 基于组播动态发现 brokers。其原理是通过组播的形式，让各个 broker 自动发现彼此并建立连接。在这种模式下，broker 在启动时会通过组播地址发送和接收发现消息，从而动态地获取可连接的 broker 地址 。

在activemq.xml中，需要对transportConnectors和networkConnectors进行相关配置修改。例如：

<broker name="foo">

<transportConnectors>

<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600" discoveryUri="multicast://default"/>

</transportConnectors>

...

<networkConnectors>

<networkConnector uri="multicast://default"/>

</networkConnectors>

</broker>

在上述配置中，transportConnector的discoveryUri属性指定了组播地址为multicast://default，表示该 broker 将通过这个组播地址进行发现；networkConnector的uri属性也设置为multicast://default，表示通过组播来建立网络连接 。这样，各个 broker 就可以通过组播自动发现彼此并建立连接，实现动态的集群配置 。动态 Broker - Cluster 配置相对灵活，不需要手动硬编码每个 broker 的地址，适合在网络环境变化频繁或需要快速扩展集群的场景中使用 。然而，组播在一些网络环境中可能会受到限制，如防火墙的阻挡等，需要在部署时特别注意网络配置 。

（三）生产环境推荐配置（Master - Slave + Broker - Cluster）

在生产环境中，单一的 Master - Slave 模式虽然能解决单点故障问题，保证消息服务的可靠性，但无法实现负载均衡；而单一的 Broker - Cluster 模式虽能实现负载均衡和消息的分布式处理，但在节点故障时可能会出现消息丢失等问题，无法满足高可用性的严格要求 。因此，将 Master - Slave 模式与 Broker - Cluster 模式结合起来，能够充分发挥两者的优势，提供更可靠、高效的消息服务。

在这种组合配置中，首先搭建 Master - Slave 集群，确保消息的可靠存储和高可用性。通过共享文件系统或 JDBC 等方式实现主从节点之间的数据同步和故障转移 。然后，在 Master - Slave 集群的基础上，搭建 Broker - Cluster，实现消息的负载均衡和分布式处理 。各个 Master 和 Slave 节点通过networkConnector相互连接，形成一个更大的集群网络 。

具体的配置步骤如下：

1. 按照前面介绍的方法，完成 Master - Slave 模式的搭建，无论是 Shared File System Master Slave 还是 JDBC Master Slave 。

1. 在每个 broker 的activemq.xml文件中，添加networkConnector配置，实现 Broker - Cluster。例如：

<networkConnectors>

<networkConnector uri="static:(tcp://broker1:61616,tcp://broker2:61617)" duplex="true"/>

</networkConnectors>

这里假设已经搭建了两个 broker 节点，分别为broker1和broker2，通过static协议配置了它们之间的网络连接，duplex="true"表示启用双向连接，即两个 broker 之间可以相互发送和接收消息 。

3. 配置客户端连接。客户端使用failover协议连接到集群，例如：

ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("failover:(tcp://broker1:61616,tcp://broker2:61617)?randomize=false&priorityBackup=true");

在这个示例中，failover协议表示当一个 broker 节点不可用时，客户端会自动切换到其他可用节点；randomize=false表示按列表顺序调用，不随机选择节点；priorityBackup=true表示当randomize为false时，优先选择第一个节点作为主节点 。

通过这种 Master - Slave + Broker - Cluster 的组合配置，既保证了消息服务的高可用性，又实现了负载均衡和分布式处理，能够满足生产环境中复杂、高并发的业务需求 。