spring boot学习笔记之操作ActiveMQ指南

前言

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合、异步消息、流量削锋等问题，实现高性能、高可用、可伸缩和最终一致性架构，是大型分布式系统不可缺少的中间件。

目前在生产环境中使用较多的消息队列有 activemq、rabbitmq、zeromq、kafka、metamq、rocketmq 等。

特性

• 异步性：将耗时的同步操作通过以发送消息的方式进行了异步化处理，减少了同步等待的时间。
• 松耦合：消息队列减少了服务之间的耦合性，不同的服务可以通过消息队列进行通信，而不用关心彼此的实现细节，只要定义好消息的格式就行。
• 分布式：通过对消费者的横向扩展，降低了消息队列阻塞的风险，以及单个消费者产生单点故障的可能性（当然消息队列本身也可以做成分布式集群）。
• 可靠性：消息队列一般会把接收到的消息存储到本地硬盘上（当消息被处理完之后，存储信息根据不同的消息队列实现，有可能将其删除），这样即使应用挂掉或者消息队列本身挂掉，消息也能够重新加载。

jms 规范

jms 即 java 消息服务（java message service）应用程序接口，是一个 java 平台中关于面向消息中间件（mom）的 api，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。java 消息服务是一个与具体平台无关的 api，绝大多数 mom 提供商都对 jms 提供支持。

jms 的消息机制有 2 种模型，一种是 point to point，表现为队列的形式，发送的消息，只能被一个接收者取走；另一种是 topic，可以被多个订阅者订阅，类似于群发。

activemq 是 jms 的一个实现。

activemq 介绍

activemq 是 apache 软件基金下的一个开源软件，它遵循 jms1.1 规范（java message service），是消息驱动中间件软件（mom）。它为企业消息传递提供高可用、出色性能、可扩展、稳定和安全保障。activemq 使用 apache 许可协议，因此，任何人都可以使用和修改它而不必反馈任何改变。

activemq 的目标是在尽可能多的平台和语言上提供一个标准的，消息驱动的应用集成。activemq 实现 jms 规范并在此之上提供大量额外的特性。activemq 支持队列和订阅两种模式的消息发送。

spring boot 提供了 activemq 组件 spring-boot-starter-activemq，用来支持 activemq 在 spring boot 体系内使用，下面我们来详细了解如何使用。

添加依赖

主要添加组件：spring-boot-starter-activemq。

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

配置文件

在 application.properties 中添加配置。

# 基于内存的 activemq
spring.activemq.in-memory=true
# 不适应连接池
spring.activemq.pool.enabled=false
 
# 独立安装的 activemq
#spring.activemq.broker-url=tcp://192.168.0.1:61616
#spring.activemq.user=admin
#spring.activemq.password=admin

在使用 activemq 时有两种使用方式，一种是使用独立安装的 activemq，在生产环境推荐使用这种；另一种是使用基于内存 activemq ，在调试阶段建议使用这种方式。

队列（queue）

队列发送的消息，只能被一个消费者接收。

创建队列

@configuration
public class mqconfig {
    @bean
    public queue queue() {
        return new activemqqueue("neo.queue");
    }
}

使用 @configuration 注解在项目启动时，定义了一个队列 queue 命名为：neo.queue。

消息生产者

创建一个消息的生产者：

@component
public class producer{
    @autowired
    private jmsmessagingtemplate jmsmessagingtemplate;
    @autowired
    private queue queue;
    public void sendqueue(string msg) {
        system.out.println("send queue msg :"+msg);
        this.jmsmessagingtemplate.convertandsend(this.queue, msg);
    }
}

jmsmessagingtemplate 是 spring 提供发送消息的工具类，使用 jmsmessagingtemplate 和创建好的 queue 对消息进行发送。

消息消费者

@component
public class consumer {
 
    @jmslistener(destination = "neo.queue")
    public void receivequeue(string text) {
        system.out.println("consumer queue msg : "+text);
    }
}

使用注解 @jmslistener(destination = "neo.queue")，表示此方法监控了名为 neo.queue 的队列。当队列 neo.queue 中有消息发送时会触发此方法的执行，text 为消息内容。

测试

创建 sampleactivemqtests 测试类，注入创建好的消息生产者。

@runwith(springrunner.class)
@springboottest
public class sampleactivemqtests {
    @autowired
    private producer producer;
    @rule
    public outputcapture outputcapture = new outputcapture();
}

outputcapture 是 spring boot 提供的一个测试类，它能捕获 system.out 和 system.err 的输出，我们可以利用这个特性来判断程序中的输出是否执行。

@test
public void sendsimplequeuemessage() throws interruptedexception {
    this.producer.sendqueue("test queue message");
    thread.sleep(1000l);
    assertthat(this.outputcapture.tostring().contains("test queue")).istrue();
}

创建测试方式，使用 producer 发送消息，为了保证容器可以接收到消息，让测试方法等待 1 秒，最后使用 outputcapture 判断是否执行成功。

测试多消费者

上面的案例只是一个生产者一个消费者，我们在模拟一个生产者和多个消费者队列的执行情况。我们复制上面的消费者 consumer 重新命名为 consumer2，并且将输出内容加上 2 的关键字，如下：

@component
public class consumer2 {
    @jmslistener(destination = "neo.queue")
    public void receivequeue(string text) {
        system.out.println("consumer2 queue msg : "+text);
    }
}

在刚才的测试类中添加一个 send100queuemessage() 方法，模式发送 100 条消息时，两个消费者是如何消费消息的。

@test
public void send100queuemessage() throws interruptedexception {
    for (int i=0;i<100;i++){
        this.producer.sendqueue("test queue message"+i);
    }
    thread.sleep(1000l);
}

控制台输出结果：

consumer queue msg : test queue message0

consumer2 queue msg : test queue message1

consumer queue msg : test queue message2

consumer2 queue msg : test queue message3

...

根据控制台输出的消息可以看出，当有多个消费者监听一个队列时，消费者会自动均衡负载的接收消息，并且每个消息只能有一个消费者所接收。

注意：控制台输出 javax.jms.jmsexception: peer (vm://localhost#1) stopped. 报错信息可以忽略，这是 info 级别的错误，是 activemq 的一个 bug。

广播（topic）

广播发送的消息，可以被多个消费者接收。

创建 topic

@configuration
public class mqconfig {
    @bean
    public topic topic() {
        return new activemqtopic("neo.topic");
    }
}

使用 @configuration 注解在项目启动时，定义了一个广播 topic 命名为：neo.topic。

消息生产者

创建一个消息的生产者：

@component
public class producer{
    @autowired
    private jmsmessagingtemplate jmsmessagingtemplate;
    @autowired
    private topic topic;
    public void sendtopic(string msg) {
        system.out.println("send topic msg :"+msg);
        this.jmsmessagingtemplate.convertandsend(this.topic, msg);
    }
}

和上面的生产者对比只是 convertandsend() 方法传入的第一个参数变成了 topic。

消息消费者

@component
public class consumer {
 
    @jmslistener(destination = "neo.topic")
    public void receivetopic(string text) {
        system.out.println("consumer topic msg : "+text);
    }
}

消费者也没有变化，只是监听的名改为上面的 neo.topic，因为模拟多个消费者，复制一份 consumer 命名为 consumer2，代码相同在输出中标明来自 consumer2。

测试

创建 sampleactivemqtests 测试类，注入创建好的消息生产者。

@test
public void sendsimpletopicmessage() throws interruptedexception {
    this.producer.sendtopic("test topic message");
    thread.sleep(1000l);
}

测试方法执行成功后，会看到控制台输出信息，如下：

send topic msg :test topic message

consumer topic msg : test topic message

consumer2 topic msg : test topic message

可以看出两个消费者都收到了发送的消息，从而验证广播（topic）是一个发送者多个消费者的模式。

同时支持队列（queue）和广播（topic）

spring boot 集成 activemq 的项目默认只支持队列或者广播中的一种，通过配置项 spring.jms.pub-sub-domain 的值来控制，true 为广播模式，false 为队列模式，默认情况下支持队列模式。

如果需要在同一项目中既支持队列模式也支持广播模式，可以通过 defaultjmslistenercontainerfactory 创建自定义的 jmslistenercontainerfactory 实例，之后在 @jmslistener 注解中通过 containerfactory 属性引用它。

分别创建两个自定义的 jmslistenercontainerfactory 实例，通过 pubsubdomain 来控制是支持队列模式还是广播模式。

@configuration
@enablejms
public class activemqconfig {
 
    @bean("queuelistenerfactory")
    public jmslistenercontainerfactory<?> queuelistenerfactory(connectionfactory connectionfactory) {
        defaultjmslistenercontainerfactory factory = new defaultjmslistenercontainerfactory();
        factory.setconnectionfactory(connectionfactory);
        factory.setpubsubdomain(false);
        return factory;
    }
 
    @bean("topiclistenerfactory")
    public jmslistenercontainerfactory<?> topiclistenerfactory(connectionfactory connectionfactory) {
        defaultjmslistenercontainerfactory factory = new defaultjmslistenercontainerfactory();
        factory.setconnectionfactory(connectionfactory);
        factory.setpubsubdomain(true);
        return factory;
    }
}

然后在消费者接收的方法中，指明使用 containerfactory 接收消息。

@component
public class consumer {
 
    @jmslistener(destination = "neo.queue", containerfactory = "queuelistenerfactory")
    public void receivequeue(string text) {
        system.out.println("consumer queue msg : "+text);
    }
 
    @jmslistener(destination = "neo.topic", containerfactory = "topiclistenerfactory")
    public void receivetopic(string text) {
        system.out.println("consumer topic msg : "+text);
    }
}

改造完成之后，再次执行队列和广播的测试方法，就会发现项目同时支持了两种类型的消息收发。

总结

消息中间件广泛应用在大型互联网架构中，利用消息中间件队列和广播各自的特性可以支持很多业务，比如群发发送短信、给单个用户发送邮件等。activemq 是一款非常流行的消息中间件，它的特点是部署简单、使用方便，比较适合中小型团队。spring boot 提供了集成 activemq 对应的组件，在 spring boot 中使用 activemq 只需要添加相关注解即可。

到此这篇关于spring boot学习笔记之操作activemq指南的文章就介绍到这了,更多相关spring boot操作activemq指南内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！