摘要：在后端开发领域，服务间的通信稳定性与效率，一直是影响系统性能的关键因素。构建大型分布式系统时，一旦出现通信延迟、消息丢失等问题，系统的稳定性和性能将受到严重影响。消息队列作为保障服务间可靠、高效通信的关键组件，其重要性不言而喻。ActiveMQ 作为一款广泛应
在后端开发领域，服务间的通信稳定性与效率，一直是影响系统性能的关键因素。构建大型分布式系统时，一旦出现通信延迟、消息丢失等问题，系统的稳定性和性能将受到严重影响。消息队列作为保障服务间可靠、高效通信的关键组件，其重要性不言而喻。ActiveMQ 作为一款广泛应用的开源消息代理，与功能强大的 Spring Boot 框架相结合，能够极大提升开发效率，优化系统架构。今天，我们就来全面且深入地探讨如何在 Spring Boot3 中整合实现 ActiveMQ 消息队列。
ActiveMQ 是 Apache 软件基金会旗下一款极为流行且功能强劲的开源消息代理。它支持 JMS、AMQP、MQTT 等多种消息传输协议，能适应不同场景下的消息通信需求。ActiveMQ 提供消息持久化功能，确保消息在系统故障时不丢失；支持消息队列、主题等多种消息模型，满足不同业务场景的消息传递要求；还具备强大的集群功能，可提升系统的可靠性和扩展性。在众多开源消息队列中，ActiveMQ 凭借广泛的应用和良好的口碑脱颖而出。
当 Spring Boot 与 ActiveMQ 整合，在构建消息驱动的应用时，能极大减少样板代码编写量。Spring Boot 提供spring-boot-starter-activemq等与 ActiveMQ 整合的相关组件，让整合过程更顺畅。随着 Spring Boot3 的发布，对消息队列的支持迎来新特性和优化，为开发者提供更多便利，也促使我们深入探究如何在 Spring Boot3 中实现与 ActiveMQ 的完美整合。
在 Spring Boot3 项目中整合 ActiveMQ，首先要在项目的pom.xml文件中添加 ActiveMQ 的起步依赖。在pom.xml的依赖部分，加入以下代码：
org.springframework.bootSpring - boot - starter - activemq添加这段依赖后，Maven 会自动下载并引入 Spring Boot 对 ActiveMQ 的支持及相关依赖包。这些依赖包包含各种工具类和接口，为后续与 ActiveMQ 的交互提供必要支持。
配置 ActiveMQ
添加完依赖后，接下来要在application.properties或application.yml文件中对 ActiveMQ 进行配置。这里以application.yml为例，配置如下：
spring:activemq:broker - url: tcp://localhost:61616上述配置中，broker - url指定了 ActiveMQ 服务器的地址。实际应用中，如果 ActiveMQ 服务器部署在其他主机上，或使用不同端口，需相应修改地址。比如，服务器 IP 是192.168.1.100，端口是61617，配置就应改为broker - url: tcp://192.168.1.100:61617。通过这一步配置，Spring Boot 应用就能确定连接 ActiveMQ 服务器的地址，建立通信通道。
配置好 ActiveMQ 后，我们需要创建一个生产者服务类，用于向 ActiveMQ 发送消息。下面是一段生产者代码示例：
import org.springframework.beans.factory.annotation.Autowired;import org.springframework.Jms.core.jmsTemplate;import org.springframework.stereotype.Service;@Servicepublic class ProducerService {@Autowiredprivate JmsTemplate jmsTemplate;public void sendMessage(String destination, String message) {jmsTemplate.send(destination, session -> session.createTextMessage(message));}}在这段代码中，ProducerService类被标记为@Service，会被 Spring 容器管理。通过@Autowired注解，将JmsTemplate注入到ProducerService中。JmsTemplate是 Spring 提供的用于操作 JMS（Java Message Service）的工具类，简化了与消息队列的交互操作。sendMessage方法接收两个参数，destination表示消息要发送到的目的地，message是要发送的具体消息内容。方法内部通过jmsTemplate.send方法将消息发送到指定目的地，并使用session.createTextMessage(message)创建一个文本类型的消息。
有了消息的发送者，自然也需要有消息的接收者。接下来创建一个消费者服务类，用于从 ActiveMQ 接收消息并进行处理。示例代码如下：
import org.springframework.jms.annotation.JmsListener;import org.springframework.stereotype.Component;@Componentpublic class QueueConsumerListener {@JmsListener(destination = "${spring.activemq.queue - name}", containerFactory = "queueListener")public void readActiveQueue(String message) {System.out.println("queue接受到：" + message);}}QueueConsumerListener类被标记为@Component，同样会被 Spring 容器管理。@JmsListener注解用于监听指定的目的地，destination属性的值通过${spring.activemq.queue - name}从配置文件中获取，containerFactory指定了使用的消息监听容器工厂。当有消息发送到指定队列时，readActiveQueue方法就会被触发，参数message就是接收到的消息内容。实际应用中，可根据业务需求进行更复杂的处理。
最后，不要忘记在 Spring Boot3 的启动类中添加@EnableJms注解，这个注解的作用是启用 JMS 功能。只有添加了这个注解，Spring Boot 应用才能正确识别和处理与 JMS 相关的配置和组件，让消息在生产者和消费者之间顺畅流动。
消息持久化
在实际应用中，消息的可靠性至关重要。ActiveMQ 支持消息持久化，这意味着即使系统出现故障或服务器重启，消息也不会丢失。当生产者发送一条持久化消息时，ActiveMQ 会将消息存储到磁盘上，只有当消费者成功接收并处理这条消息后，ActiveMQ 才会将其从存储中删除。要实现消息持久化，在生产者发送消息时，需要设置消息的持久化属性。例如：
public void sendPersistentMessage(String destination, String message) {MessagePostProcessor messagePostProcessor = new MessagePostProcessor {@Overridepublic Message postProcessMessage(Message message) throws JMSException {message.setJMSDeliveryMode(DeliveryMode.PERSISTENT);return message;}};jmsTemplate.send(destination, session -> session.createTextMessage(message), messagePostProcessor);}在这段代码中，通过MessagePostProcessor设置消息的JMSDeliveryMode为DeliveryMode.PERSISTENT，表示这是一条持久化消息。这样，ActiveMQ 接收到消息后，会将其持久化存储，确保消息的可靠性。
集群部署
随着业务量的增长，单个 ActiveMQ 服务器可能无法满足系统的性能和可靠性要求。此时，就需要对 ActiveMQ 进行集群部署。ActiveMQ 集群可以通过基于网络共享存储的主从模式，或基于复制协议的多主模式等多种方式实现。在集群环境下，多个 ActiveMQ 节点可以共同分担消息处理压力，提高系统的吞吐量和响应速度。同时，当某个节点出现故障时，其他节点可以继续提供服务，保证系统的高可用性。在 Spring Boot 项目中配置 ActiveMQ 集群，需要在application.yml中对broker - url进行相应修改，例如：
spring:activemq:broker - url: failover:(tcp://node1:61616,tcp://node2:61616,tcp://node3:61616)上述配置中，failover协议表示在多个节点之间进行故障转移。当连接到某个节点失败时，会自动尝试连接其他节点，从而保证系统的稳定性。
消息过滤
在复杂的业务场景中，消费者可能只需要接收特定条件的消息。ActiveMQ 支持消息过滤功能，生产者发送消息时，可以为消息设置一些属性，消费者接收消息时，可以根据这些属性进行过滤。例如，生产者发送消息时设置一个自定义属性：
public void sendMessageWithProperty(String destination, String message) {MessagePostProcessor messagePostProcessor = new MessagePostProcessor {@Overridepublic Message postProcessMessage(Message message) throws JMSException {message.setStringProperty("messageType", "important");return message;}};jmsTemplate.send(destination, session -> session.createTextMessage(message), messagePostProcessor);}消费者接收消息时，可以通过@JmsListener注解的selector属性进行过滤：
@Componentpublic class FilteredConsumerListener {@JmsListener(destination = "${spring.activemq.queue - name}", selector = "messageType = 'important'", containerFactory = "queueListener")public void receiveFilteredMessage(String message) {System.out.println("接收到重要消息：" + message);}}在这个例子中，只有消息的messageType属性值为important的消息才会被FilteredConsumerListener接收和处理，这样可以帮助消费者精准获取所需信息，提高系统的效率。
通过以上详细步骤和深入拓展，我们全面了解了如何在 Spring Boot3 中整合实现 ActiveMQ 消息队列，以及其在实际应用中的高级场景。实际项目开发中，开发者可根据具体业务需求，灵活调整配置和代码，充分发挥 Spring Boot 与 ActiveMQ 整合的优势。
整合过程中，可能会遇到依赖冲突、配置错误等问题。如果你在实践过程中遇到难题，或有独特的优化经验和更好的实践方案，欢迎在评论区分享交流。让我们一起共同进步，打造更高效、稳定的分布式系统。希望这篇文章能为你在 Spring Boot3 中整合 ActiveMQ 提供有力帮助，提升开发效率，解决实际困扰。
来源：从程序员到架构师一点号