Spring Boot整合ActiveMQ Artemis指南-java教程-PHP中文网

spring boot整合activemq artemis的核心在于利用其自动化配置和依赖管理简化jms集成。1. 添加spring-boot-starter-activemq依赖实现快速接入；2. 在application.properties或yml中配置broker-url、用户信息等；3. 使用jmstemplate发送消息，支持字符串和对象的序列化；4. 通过@jmslistener注解监听队列或主题，可区分队列与主题的监听器；5. 可自定义jmslistenercontainerfactory以支持不同消息域和转换器；6. 配置连接池提升性能，如使用jmspoolconnectionfactory；7. 自定义消息转换器如mappingjackson2messageconverter处理json；8. 启用事务管理确保消息与业务操作的一致性；9. 设置errorhandler实现错误处理与死信队列机制；10. 生产环境需考虑持久化、高可用部署（如shared store或replication ha）、安全认证授权、监控告警及资源调优等关键点，以保障系统的稳定性和可靠性。

Spring Boot整合ActiveMQ Artemis指南

Spring Boot整合ActiveMQ Artemis，在我看来，核心就是利用Spring Boot的自动化配置和依赖管理能力，高效、便捷地将Artemis这个强大而灵活的消息中间件融入到你的应用中。它极大地简化了JMS（Java Message Service）的配置和使用，让开发者能更专注于业务逻辑，而非繁琐的底层集成细节。

Spring Boot与ActiveMQ Artemis的整合，其实远比你想象的要直接。它就像Spring Boot一贯的风格，提供了开箱即用的解决方案。

首先，你需要确保你的项目依赖中包含了ActiveMQ的Starter。通常，spring-boot-starter-activemq就足够了，因为它内部已经包含了Artemis的客户端库。如果你用Maven，那就像这样：

<dependency><groupid>org.springframework.boot</groupid><artifactid>spring-boot-starter-activemq</artifactid></dependency>

接着，配置ActiveMQ Artemis的连接信息。这部分通常在application.properties或application.yml里搞定。如果你是在本地跑一个Artemis实例，或者连接远程Broker，无非就是指定地址、端口、用户名和密码。

# application.properties
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
# 开启JMS消息池，这是个好习惯，能提升性能
spring.jms.cache.connection-factory=true

发送消息，Spring Boot会自动配置一个JmsTemplate bean，你直接注入使用就行。这简直是福利，省去了手动创建ConnectionFactory、JmsTemplate的麻烦。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {

    private final JmsTemplate jmsTemplate;

    @Autowired
    public MessageProducer(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    public void sendMessage(String destination, String message) {
        System.out.println("Sending message to " + destination + ": " + message);
        jmsTemplate.convertAndSend(destination, message);
    }

    public void sendObjectMessage(String destination, MyCustomObject obj) {
        System.out.println("Sending object message to " + destination + ": " + obj);
        // Spring Boot默认支持序列化Java对象，但更推荐使用JSON等格式
        jmsTemplate.convertAndSend(destination, obj);
    }
}

接收消息，Spring Boot提供了@JmsListener注解，这真是太方便了。你只需要在一个方法上加上这个注解，指定监听的队列或主题，Spring Boot就会自动帮你处理消息的消费。

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

@Component
public class MessageConsumer {

    @JmsListener(destination = "my.queue")
    public void receiveQueueMessage(String message) {
        System.out.println("Received queue message: " + message);
    }

    @JmsListener(destination = "my.topic", containerFactory = "jmsTopicListenerContainerFactory")
    public void receiveTopicMessage(String message) {
        System.out.println("Received topic message: " + message);
    }

    // 如果要接收对象消息，并且想用JSON转换器
    @JmsListener(destination = "my.object.queue", containerFactory = "jmsQueueListenerContainerFactory")
    public void receiveObjectMessage(MyCustomObject obj) {
        System.out.println("Received object message: " + obj);
    }
}

注意，如果你需要区分队列和主题的监听器，可能需要额外配置一个JmsListenerContainerFactory，比如用于主题的：

import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import javax.jms.ConnectionFactory;

@Configuration
public class JmsConfig {

    @Bean
    public JmsListenerContainerFactory> jmsTopicListenerContainerFactory(
            ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setPubSubDomain(true); // 开启主题模式
        return factory;
    }

    @Bean
    public JmsListenerContainerFactory> jmsQueueListenerContainerFactory(
            ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setPubSubDomain(false); // 队列模式是默认的，但明确设置也无妨
        // factory.setMessageConverter(jacksonJmsMessageConverter()); // 如果需要自定义消息转换器
        return factory;
    }

    // 如果需要发送和接收JSON对象，可以配置一个消息转换器
    // @Bean
    // public MessageConverter jacksonJmsMessageConverter() {
    //     MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
    //     converter.setTargetType(MessageType.TEXT);
    //     converter.setTypeIdPropertyName("_type");
    //     return converter;
    // }
}

ActiveMQ Artemis为何在Spring Boot生态中如此受欢迎？

在我看来，ActiveMQ Artemis在Spring Boot生态中受欢迎并非偶然，它有着几个非常实际的优势，让其在众多消息队列中脱颖而出。

首先，性能与资源消耗。Artemis是基于Netty构建的，它的异步I/O模型使得其在处理高并发消息时表现出色，延迟低、吞吐量高。相比起一些“老牌”消息队列，Artemis在资源占用上更为轻量，这对于微服务架构下资源敏感的应用来说，无疑是个巨大的加分项。我个人在实际项目中对比过，Artemis在同等负载下的表现确实令人满意。

其次，多协议支持。Artemis不仅仅支持JMS，它还原生支持AMQP、MQTT、STOMP、OpenWire等多种协议。这意味着你的系统可以非常灵活地与不同技术栈、不同客户端进行集成，这在复杂的企业环境中尤其重要。比如，你可能有一个IoT设备通过MQTT发送数据，同时后端服务通过JMS消费，Artemis能完美地充当这个桥梁。这种灵活性，让它不仅仅是一个JMS Broker。

再者，嵌入式部署能力。Artemis可以非常方便地作为嵌入式Broker运行在你的Spring Boot应用内部。这对于开发测试环境、或者一些对消息可靠性要求没那么高、但又需要消息队列功能的轻量级应用来说，简直是福音。省去了独立部署的麻烦，开发体验也流畅很多。虽然生产环境通常还是独立部署，但这个特性在某些场景下确实很有用。

最后，社区活跃度与演进。Artemis是Apache ActiveMQ的下一代产品，它继承了ActiveMQ的优点，同时进行了大量的优化和改进。社区活跃，文档丰富，遇到问题也比较容易找到解决方案。它一直在积极演进，不断引入新的特性和优化，这给了开发者信心。

如何在Spring Boot中精细化配置ActiveMQ Artemis的连接与消息处理？

当你的应用走向生产环境，或者对消息的可靠性、性能有更高要求时，仅仅依靠默认配置是远远不够的。Spring Boot虽然提供了极大的便利，但它也保留了让你深入定制的能力。

一个常见的需求是连接池的优化。Spring Boot默认会使用CachingConnectionFactory，它会缓存Connection、Session和MessageProducer。但你可以进一步配置其缓存大小、会话模式等。

Miller CSV TSV JSON 数据处理器

Miller (mlr) 是一个命令行工具，用于查询、整形和重新格式化名称索引数据，如 CSV、TSV、JSON 和 JSON Lines。它将 awk、sed、cut、join 和 sort 的功能整合到一个专为结构化数据处理而构建的单一工具中。

下载

# application.properties
# 连接工厂缓存设置
spring.jms.cache.connection-factory=true
spring.jms.cache.session-cache-size=10 # 每个连接缓存的会话数量

如果你需要更细粒度的控制，比如使用PooledConnectionFactory，你可能需要手动配置JmsConnectionFactory和JmsTemplate：

import org.apache.activemq.artemis.jms.client.ActiveMQJMSConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQXAConnectionFactory; // 如果需要XA事务
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.core.JmsTemplate;

@Configuration
public class CustomJmsConfig {

    @Bean
    public ActiveMQJMSConnectionFactory artemisConnectionFactory() {
        // 直接创建Artemis的ConnectionFactory
        return new ActiveMQJMSConnectionFactory("tcp://localhost:61616", "admin", "admin");
    }

    @Bean(destroyMethod = "stop")
    public JmsPoolConnectionFactory pooledConnectionFactory(ActiveMQJMSConnectionFactory artemisConnectionFactory) {
        JmsPoolConnectionFactory pool = new JmsPoolConnectionFactory();
        pool.setConnectionFactory(artemisConnectionFactory);
        pool.setMaxConnections(10); // 最大连接数
        pool.setBlockIfSessionPoolIsFull(true);
        pool.setBlockIfSessionPoolIsFullTimeout(5000); // 阻塞超时时间
        return pool;
    }

    @Bean
    public JmsTemplate jmsTemplate(JmsPoolConnectionFactory pooledConnectionFactory) {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(pooledConnectionFactory);
        // template.setMessageConverter(jacksonJmsMessageConverter()); // 如果需要自定义消息转换器
        return template;
    }
}

消息转换器是另一个需要关注的地方。默认情况下，JMS可以发送和接收String、Map、Bytes以及可序列化的Java对象。但在微服务和异构系统集成中，JSON或XML是更常见的选择。你可以配置MappingJackson2MessageConverter来处理JSON：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.support.converter.MappingJackson2MessageConverter;
import org.springframework.jms.support.converter.MessageType;

@Configuration
public class MessageConverterConfig {

    @Bean
    public MappingJackson2MessageConverter jacksonJmsMessageConverter() {
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setTargetType(MessageType.TEXT); // 将Java对象转换为文本消息（JSON字符串）
        converter.setTypeIdPropertyName("_type"); // 在消息头中添加类型信息，方便反序列化
        return converter;
    }
}
// 别忘了在JmsListenerContainerFactory或JmsTemplate中设置这个Converter

事务管理也是个核心。JMS支持本地事务和XA事务。对于简单的“发送即确认”或“接收即确认”场景，可以不使用事务。但如果涉及数据库操作和消息发送的原子性，就需要事务了。Spring的@Transactional注解可以与JMS很好地配合，但前提是你的JmsListenerContainerFactory配置了事务管理器。

// 在JmsConfig中，确保JmsListenerContainerFactory支持事务
// factory.setSessionTransacted(true); // 开启JMS会话事务
// 结合Spring的PlatformTransactionManager
// @Transactional 注解就可以生效了

最后，错误处理。消息消费过程中难免会遇到异常。Spring Boot的@JmsListener默认会重试，但你可能需要更细致的控制，比如死信队列（DLQ）或者自定义错误处理器。你可以为JmsListenerContainerFactory设置一个ErrorHandler：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.util.ErrorHandler;

@Configuration
public class JmsErrorHandlerConfig {

    @Bean
    public ErrorHandler myJmsErrorHandler() {
        return t -> {
            System.err.println("JMS message processing error: " + t.getMessage());
            // 这里可以记录日志、发送告警、或者将消息转发到死信队列
            // 避免抛出异常导致消息无限重试
        };
    }

    // 在JmsListenerContainerFactory中设置这个ErrorHandler
    // factory.setErrorHandler(myJmsErrorHandler());
}

这些精细化配置能让你更好地掌控消息的生命周期和处理流程，确保系统在复杂场景下的稳定性和可靠性。

ActiveMQ Artemis在生产环境部署中的关键考量点有哪些？

将ActiveMQ Artemis从开发测试环境推向生产，绝不是简单地把代码部署上去就完事。生产环境的稳定性和高可用性是重中之重，这要求我们对Artemis的部署策略、持久化、集群、安全和监控等方面有深入的理解和规划。

首先，持久化。Artemis默认会使用日志文件（journal）进行消息持久化，这通常非常高效。但你需要考虑日志文件的存储位置、大小以及如何防止磁盘空间耗尽。对于关键业务消息，必须确保它们被正确持久化。你也可以配置Artemis使用数据库（JDBC）进行持久化，但这通常会带来性能开销，除非你有特殊的集成需求。我的建议是，优先考虑Artemis原生的Journal，它的性能表现通常更优。

其次，高可用性（HA）和集群。这是生产环境的标配。Artemis提供了几种HA模式：

共享存储（Shared Store）HA：多个Broker实例共享同一个持久化存储（比如NFS、SAN）。一个Broker作为主节点活跃，其他作为备份。当主节点故障时，备份节点会接管。这种模式实现简单，但共享存储可能成为单点故障。
复制（Replication）HA：这是我个人更推荐的模式。Broker之间通过网络同步数据，没有共享存储依赖。一个Broker是主节点，其他是复制节点。主节点故障时，复制节点可以提升为主节点。这种模式提供了更高的可靠性和数据安全性，但网络延迟会影响同步性能。

选择哪种模式取决于你的业务对数据一致性、恢复时间目标（RTO）和恢复点目标（RPO）的要求。

再者，安全性。生产环境的Artemis必须配置认证和授权。默认的admin/admin账户是绝对不能用的。你需要配置用户、角色，并为队列和主题设置访问权限。Artemis支持多种认证机制，包括基于文件、LDAP、Kerberos等。对于Spring Boot应用，连接Artemis时也需要使用配置好的安全凭证。

<!-- Artemis broker.xml 配置示例，用于安全 -->
<security-settings><security-setting match="#"><permission type="createNonDurableQueue" roles="admin,dev"></permission><permission type="deleteNonDurableQueue" roles="admin,dev"></permission><permission type="createDurableQueue" roles="admin"></permission><permission type="deleteDurableQueue" roles="admin"></permission><permission type="send" roles="admin,producer"></permission><permission type="consume" roles="admin,consumer"></permission><permission type="manage" roles="admin"></permission></security-setting></security-settings>

还有监控与告警。你需要能够实时监控Artemis的运行状态，包括连接数、消息堆积量、内存使用、CPU负载等。Artemis通过JMX提供了丰富的监控指标，你可以使用JConsole、VisualVM等工具连接，或者集成到Prometheus、Grafana等监控系统中。设置合理的告警阈值，确保在问题发生时能及时收到通知。

最后，资源管理。这包括JVM参数调优（内存、GC）、文件描述符限制、网络配置等。对于高吞吐量的Artemis实例，适当增加文件描述符限制和调整JVM堆大小是很有必要的。

生产环境的部署是一个系统工程，需要综合考虑性能、可靠性、安全和可维护性。没有一劳永逸的方案，但遵循这些关键考量点，能让你少走很多弯路。