SpringBoot整合ActiveMQ（入门）

这是一篇博客原文。

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

ActiveMQ是什么？

1. ActiveMQ是消息队列技术，为解决高并发问题而生
2. ActiveMQ生产者消费者模型（生产者和消费者可以跨平台、跨系统）
3. ActiveMQ支持如下两种消息传输方式

• 点对点模式，生产者生产了一个消息，只能由一个消费者进行消费
• 发布/订阅模式，生产者生产了一个消息，可以由多个消费者进行

SpringBoot整合ActiveMQ

1. ActiveMQ下载启动

可直接进入官网下载完成。 http://activemq.apache.org/download-archives.html

  如果能成功访问http://localhost:8161/admin（用户名和密码默认为admin），则启动成功。

2.  创建两个springboot项目，分别作为消息提供者（provider）和消费者（consumer），添加依赖

<dependency>     <groupId>org.springframework.boot</groupId>     <artifactId>spring-boot-starter-activemq</artifactId> </dependency> <!--消息队列连接池--> <dependency>     <groupId>org.apache.activemq</groupId>     <artifactId>activemq-pool</artifactId>     <version>5.15.0</version> </dependency>

3.  在两个项目中的application.properires配置消息队列，并在启动类添加@EnableJms开启消息队列。

application.properties

# failover:(tcp://localhost:61616,tcp://localhost:61617) # tcp://localhost:61616 spring.activemq.broker-url=tcp://localhost:61616 #true 表示使用内置的MQ，false则连接服务器 spring.activemq.in-memory=false #true表示使用连接池；false时，每发送一条数据创建一个连接 spring.activemq.pool.enabled=true #连接池最大连接数 spring.activemq.pool.max-connections=10 #空闲的连接过期时间，默认为30秒 spring.activemq.pool.idle-timeout=30000 #强制的连接过期时间，与idleTimeout的区别在于：idleTimeout是在连接空闲一段时间失效，而expiryTimeout不管当前连接的情况，只要达到指定时间就失效。默认为0，never spring.activemq.pool.expiry-timeout=0

这里采用的是key---value键值对方式配置。  也可以使用yaml格式配置（此格式为官方推荐格式）

启动类(provider)，consumer同样

@SpringBootApplication @EnableJms //启动消息队列 public class ProviderApplication { 	public static void main(String[] args) { 		SpringApplication.run(ProviderApplication.class, args); 	} }

4.  prover项目结构图

BeanConfig定义消息队列

import javax.jms.Queue; import org.apache.activemq.command.ActiveMQQueue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration;   /*  * @author ww  * @date 2021/3/23  11:55  */ @Configuration public class BeanConfig {       //定义存放消息的队列     @Bean     public Queue queue() {         return new ActiveMQQueue("ActiveMQQueue");     } }

ProviderController

import javax.jms.Queue; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController;   /*  * @author ww  * @date   *  */ @RestController public class ProviderController {       //注入存放消息的队列，用于下列方法一     @Autowired     private Queue queue;       //注入springboot封装的工具类     @Autowired     private JmsMessagingTemplate jmsMessagingTemplate;       @RequestMapping("send")     public void send(String name) {         //方法一：添加消息到消息队列         jmsMessagingTemplate.convertAndSend(queue, name);         //方法二：这种方式不需要手动创建queue，系统会自行创建名为test的队列         //jmsMessagingTemplate.convertAndSend("test", name);     } }

5.  启动provider，向消息队列添加数据，本次添加5条数据

这里使用postman工具模拟请求发送。

查看http://localhost:8161/admin/queues.jsp 如下

• Number Of Pending Messages：消息队列中待处理的消息
• Number Of Consumers：消费者的数量
• Messages Enqueued：累计进入过消息队列的总量
• Messages Dequeued：累计消费过的消息总量

 6. consumer项目结构图

application.properties 和 ConsumerApplication 同 provider类似，如下为不同的ConsumerService：

import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.annotation.JmsListener; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.stereotype.Component;   /*  * @author ww  * @date 2021/3/23  11:58  *  */ @Component public class ConsumerService {       @Autowired     private JmsMessagingTemplate jmsMessagingTemplate;       // 使用JmsListener配置消费者监听的队列，其中name是接收到的消息     @JmsListener(destination = "ActiveMQQueue")     // SendTo 会将此方法返回的数据, 写入到 OutQueue 中去.     @SendTo("SQueue")     public String handleMessage(String name) {         System.out.println("成功接受Name" + name);         return "成功接受Name" + name;     } }

 7. 启动consumer，控制台输出如下

 消息接收成功，查看http://localhost:8161/admin/queues.jsp ，如下图所示，消息队列中不再有未处理的消息，由于consumer的启动，消费者的数量为1，Messages Dequeued（累计消费过的消息总量）的数值也变成了5；另外消费者接收到5条消息处理后，返回到OutQueue 5条消息，下图可以看出来。