Spring Boot 中集成 ActiveMQ

ActiveMQ

        ActiveMQ 是 Apache 推出的一个能力强劲的开源消息总线。ActiveMQ 完全支持 JMS 1.1 和 J2EE 1.4 规范，尽管 JMS 规范出台已经很久了，JMS 在当今的 Java EE 应用中仍扮演着特殊的地位。ActiveMQ 主要用在异步消息的处理上，所谓异步消息即消息发送者无需等待消息接收者的处理以及返回，甚至无需关心消息是否发送成功。

        异步消息主要有两种形式，队列（Queue）和主题（Topic），队列用于点对点形式的消息通信，主题用于发布/订阅式的消息通信。本文主要来学习一下在 Spring Boot 中如何使用这两种形式的消息。

ActiveMQ 集成

1. 依赖导入和配置

在 Spring Boot 中集成 ActiveMQ 需要导入如下 starter 依赖：

然后在 application.yml 配置文件中，对 ActiveMQ 做下配置：

2. Queue 和 Topic 的创建

首先需要创建两种消息 Queue 和 Topic，我们放到 ActiveMqConfig 中创建，如下：

可以看出 Queue 和 Topic 两种消息，分别使用 new ActiveMQQueue 和 new ActiveMQTopic 来创建，后面的参数写上对应的消息名称即可。这样，在其他地方就可以直接将这两种消息作为组件注入进来。

3. 消息的发送接口

在 Spring Boot 中，我们只要注入 JmsMessagingTemplate 模板即可快速发送消息，如下：

convertAndSend 方法中第一个参数是消息发送的目的地，第二个参数是具体的消息内容。

4. 点对点消息生产与消费

1）点对点消息的生产

消息的生产，我们放到 Controller 中来做，由于上面已经生成了 Queue 消息的组件，所以在 Controller 中我们直接注入进来即可。之后调用上文的消息发送方法 sendMessage 即可成功生产一条消息。

2）点对点消息的消费

点对点消息的消费很简单，只要我们指定目的地即可，JMS 监听器一直在监听是否有消息过来，如果有，则消费。

可以看出，使用 @JmsListener 注解来指定要监听的目的地，在消息接收方法内部，我们可以根据具体的业务需求做相应的逻辑处理。

5. 发布/订阅消息的生产和消费

1）发布/订阅消息的生产

和点对点消息一样，我们注入 Topic 并调用 producer 的 sendMessage 方法即可发送订阅消息，如下，不再赘述：

2）发布/订阅消息的消费

发布/订阅消息的消费和点对点不同，订阅消息支持多个消费者一起消费。其次，Spring Boot 中默认的是点对点消息，所以在使用 Topic 时，会不起作用，我们需要在配置文件 application.yml 中添加一个配置：

该配置是 false 的话，则为点对点消息，也是 Spring Boot 默认的。这样确实可以解决问题，但这样配置的话，上面提到的点对点消息又不能正常消费了。二者不可兼得，所以这并非一个好的解决办法。

我们来看一个比较好的解决办法。定义一个工厂，@JmsListener 注解默认只接收 Queue 消息，如果要接收 Topic 消息，需要设置一下 containerFactory。我们还在上面那个 ActiveMqConfig 配置类中添加：

@Configuration

public class ActiveMqConfig {

    @Bean

    public JmsListenerContainerFactory topicListenerContainer(ConnectionFactory connectionFactory) {

DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();

factory.setConnectionFactory(connectionFactory);

        // 相当于在 application.yml 中配置 spring.jms.pub-sub-domain=true

factory.setPubSubDomain(true);

        return factory;

}

}

经过这样的配置之后，在 @JmsListener 注解中指定这个容器工厂即可消费 Topic 消息。如下：

@Service

public class TopicConsumer1 {

    @JmsListener(destination = ActiveMqConfig.TOPIC_NAME, containerFactory = "topicListenerContainer")

    public void receiveTopicMsg(String msg) {

System.out.println("收到的消息为：" + msg);

}

}

指定 containerFactory 属性为上面我们自己配置的 topicListenerContainer 即可。

总结

本文主要介绍了 JMS 和 ActiveMQ 的相关概念及安装与启动，并详细分析了 Spring Boot 中点对点消息和发布/订阅消息两种方式的配置、消息生产和消费方式。ActiveMQ 是能力强劲的开源消息总线，在异步消息的处理上很有用，希望大家好好消化一下。