第 4-5 课：使⽤ Spring Boot 操作 ActiveMQ

Myth丶恋晨 2023-06-28 15:28 8阅读 0赞

消息队列中间件是分布式系统中重要的组件，主要解决应⽤耦合、异步消息、流量削锋等问题，实现⾼性

能、⾼可⽤、可伸缩和最终⼀致性架构，是⼤型分布式系统不可缺少的中间件。

⽬前在⽣产环境中使⽤较多的消息队列有  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");
     }
    }

使⽤  @Confifiguration  注解在项⽬启动时，定义了⼀个队列  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
    ...

根据控制台输出的消息可以看出，当有多个消费者监听⼀个队列时，消费者会⾃动均衡负载的接收消息，并

且每个消息只能有⼀个消费者所接收。 GitChat

注意：控制台输出  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");
     }
    }

使⽤  @Confifiguration  注解在项⽬启动时，定义了⼀个⼴播  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  来控制是⽀持队列模式还

是⼴播模式。 GitChat

@Configuration
    @EnableJms
    public class ActiveMQConfig {
     @Bean("queueListenerFactory")
     public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory c
    onnectionFactory) {
     DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContain
    erFactory();
     factory.setConnectionFactory(connectionFactory);
     factory.setPubSubDomain(false);
     return factory;
     }
     @Bean("topicListenerFactory")
     public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory c
    onnectionFactory) {
     DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContain
    erFactory();
     factory.setConnectionFactory(connectionFactory);
     factory.setPubSubDomain(true);
     return factory;
     }
    }

然后在消费者接收的⽅法中，指明使⽤  containerFactory  接收消息。

@Component
    public class Consumer {
     @JmsListener(destination = "neo.queue", containerFactory = "queueListenerFacto
    ry")
     public void receiveQueue(String text) {
     System.out.println("Consumer queue msg : "+text);
     }
     @JmsListener(destination = "neo.topic", containerFactory = "topicListenerFacto
    ry")
     public void receiveTopic(String text) {
     System.out.println("Consumer topic msg : "+text);
     }
    }

改造完成之后，再次执⾏队列和⼴播的测试⽅法，就会发现项⽬同时⽀持了两种类型的消息收发。

### 总结 ###

消息中间件⼴泛应⽤在⼤型互联⽹架构中，利⽤消息中间件队列和⼴播各⾃的特性可以⽀持很多业务，⽐如

群发发送短信、给单个⽤户发送邮件等。 ActiveMQ  是⼀款⾮常流⾏的消息中间件，它的特点是部署简单、使

⽤⽅便，⽐较适合中⼩型团队。 Spring Boot  提供了集成  ActiveMQ  对应的组件，在  Spring Boot  中使⽤

ActiveMQ  只需要添加相关注解即可。

[点击这⾥下载源码][Link 1]

[Link 1]: https://github.com/ityouknow/spring-boot-leaning/tree/gitbook_column2.0