这是一篇博客原文。
消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题。实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。消息形式支持点对点和订阅-发布。
ActiveMQ是什么？
- ActiveMQ是消息队列技术，为解决高并发问题而生
- ActiveMQ生产者消费者模型（生产者和消费者可以跨平台、跨系统）
- 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条消息，下图可以看出来。
