SpringBoot集成ActiveMQ的实战全过程
⽬录
前⾔
JMS规范
ActiveMQ介绍
Spring Boot集成ActiveMQ
创建项⽬并引⼊依赖
配置⽂件
订阅模式实例
同时⽀持两种形式
其他事项
参考⽂章：
总结
前⾔
在项⽬开发的过程中我们经常会遇到类似的业务场景：⽤户申请提现，后台进⾏账务处理、发送提现短信、调⽤银⾏打款通道。 在这个过程中调⽤三⽅通道（短信或银⾏通道）都⽐较耗时，同时账务处理可能也是由专门的账务系统进⾏处理。那么，为了提⾼并发和相应速度，后⾯的三个操作都可以通过异步进⾏处理。这就⽤到了消息队列。 消息队列中间件是分布式系统中重要的组件，主要解决应⽤耦合、异步消息、流量削锋等问题，实现⾼性能、⾼可⽤、可伸缩和最终⼀致性架构，是⼤型分布式系统不可缺少的中间件。
市⾯上⽐较常见的消息队列有：ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ。
在Spring Boot的starter中专门集成了ActiveMQ，因此，本篇⽂章我们就来讲讲对ActiveMQ的集成。
JMS规范
JMS即Java消息服务（Java Message Service）应⽤程序接⼝，是⼀个Java平台中关于⾯向消息中间件（MOM）的API，⽤于在两个应⽤程序之间，或分布式系统中发送消息，进⾏异步通信。Java消息服务是⼀个与具体平台⽆关的API，绝⼤多数MOM提供商都对JMS提供⽀持。
JMS的消息机制有2种模型，⼀种是队列的形式（Point to Point—）发送的消息只能被⼀个消费者消费；⼀种是订阅（Topic）模式，可以被多个订阅者订阅，订阅者都会接收到同样的消息。 ⽽ActiveMQ就是对JMS的实现之⼀。
ActiveMQ介绍
ActiveMQ是⼀种开源的基于JMS（Java Message Servie）规范的⼀种消息中间件的实现，ActiveMQ的设计⽬标是提供标准的、⾯向消息的、能够跨越多语⾔和多系统的应⽤集成消息通信中间件。
它为企业应⽤中消息传递提供⾼可⽤、出⾊性能、可扩展、稳定和安全保障。
ActiveMQ实现JMS规范并在此之上提供⼤量额外的特性。ActiveMQ⽀持队列和订阅两种模式的消息发送。
AcitveMQ的数据传送流程如下图：
ActiveMQ的两种消息传递类型：
（1）点对点传输，即⼀个⽣产者对应⼀个消费者，⽣产者向broke推送数据，数据存储在broke的⼀个队列中，当消费者接受该条队列⾥的数据。
（2）基于发布/订阅模式的传输，即根据订阅话题来接收相应数据，⼀个⽣产者可向多个消费者推送数据，与MQTT协议的实现是类似的。
两种消息传递类型的不同，点对点传输消费者可以接收到在连接之前⽣产者所推送的数据，⽽基于发布/订阅模式的传输⽅式消费者只能接收到连接之后⽣产者推送的数据。
Spring Boot集成ActiveMQ
Spring Boot针对ActiveMQ专门提供了spring-boot-starter-activemq，⽤来⽀持ActiveMQ在Spring Boot的⾃动集成配置。在此基础上我们可以很轻易的进⾏集成和使⽤。
创建项⽬并引⼊依赖
创建标准的Spring Boot项⽬，并在项⽬中引⼊以下依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>颗粒冷却塔
此时如果不需要web或其他相关处理，只引⼊该依赖即可。如果使⽤pool的话, 就需要在pom中加⼊以下依赖：
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
</dependency>
配置⽂件
在application.properties中添加如下配置：
# 基于内存的ActiveMQ
spring.activemq.in-memory=true
# 不使⽤连接池，如果使⽤连接池还需在pom中添加activemq-pool的依赖
spring.abled=false
# 独⽴安装的ActiveMQ
#spring.activemq.broker-url=tcp://127.0.0.1:61616
#spring.activemq.user=admin
#spring.activemq.password=admin
上述配置中有两套配置，Spring Boot⽀持基于内存ActiveMQ和基于独⽴安装的ActiveMQ。正常请求基于内存的形式是为了⽅便测试⽽使⽤，基于独⽴安装的形式才是真正⽤于⽣产环境。此处为了讲解功能，⽅便测试，采⽤基于内存的形式。
队列模式实例
⾸先，我们来实现基于队列（Queue）形式的实现。这⾥需要⽤到两个类ActiveMQQueue和JmsMessagingTemplate。前者是由ActiveMQ对javax.jms.Queue的接⼝实现。后者为Spring提供发送消息的⼯具类，结合Queue对消息进⾏发送。
JmsMessagingTemplate默认已经被实例化，直接拿来使⽤即可。⽽ActiveMQQueue则需要我们进⾏实例化，并传⼊消息队列的名称。
@Configuration
public class MyMqConfig {
@Bean
public Queue queue() {
return new ActiveMQQueue("sms.queue");
}
}
Spring Boot中很常规的实例化操作，不再赘述。当实例化完ActiveMQQueue之后，我们的队列便创建完成，下⾯创建对应的⽣产者和消费者。
⽣产者对应代码如下：
@Component
public class Producer {
@Resource
private JmsMessagingTemplate jmsMessagingTemplate;
@Resource
private Queue queue;
public void sendMsg(String msg) {
System.out.println("发送消息内容 :" + msg);
vertAndSend(this.queue, msg);
}
}
此处⽤到JmsMessagingTemplate和Queue，上⾯已经提到，这两个类都已经完成了初始化。消费者对应的配置如下：
@Component
public class Consumer {
@JmsListener(destination = "sms.queue")
public void receiveMsg(String text) {
System.out.println("接收到消息 : "+text);
}
}
Spring提供了注解式端点：使⽤@JmsListener。使⽤@JmsListener托管bean的带注释⽅法对其进⾏订阅。在Java8中，@JmsListener是⼀个可重复的注解，可以关联多个JMS destinations到同⼀个⽅法中。⽽在Java 6和7中，可以使⽤
@JmsListeners注解。
其中destination指定监控的消息队列名称为“sms.queue”。当队列sms.queue中有消息发送时会触发此⽅法的执⾏，text为消息内容。
上⾯完成了队列初始化、⽣产者和消费者代码的编写，下⾯通过单元测试来验证是否能够正确发送和处理消息。
@RunWith(SpringRunner.class)
@SpringBootTest
public class ActiveMqTests {
@Autowired
private Producer producer;
@Test
public void sendSimpleQueueMessage() {
this.producer.sendMsg("提现200.00元");
}
}
正火工艺执⾏单元测试，会发现在⽇志中打印如下信息：
发送消息内容 ：提现200.00元
接收到消息 : 提现200.00元
微型压力传感器芯片
说明消息可以正常发送和接收。如果是基于内存模式，在执⾏单元测试时会打印出“javax.jms.JMSException: peer
(vm://localhost#1) stopped.”异常⽇志，这是Info级别的错误，是ActiveMQ的⼀个bug。
订阅模式实例
⼴播发送的消息，可以被多个消费者接收。这⾥我们就在原有的基础上进⾏⼴播消息的添加。
⾸先，Spring Boot集成ActiveMQ时默认只⽀持队列或者⼴播之⼀，通过配置项spring.jms.pub-sub-domain来指定，true 为⼴播模式，false为队列模式，默认情况下⽀持队列模式。
此时要使⽤⼴播模式，则需在配置⽂件中添加如下配置：过氧化氢酶活性测定
spring.jms.pub-sub-domain=true
需要注意的是，此时队列模式不可正常⼯作。
然后在MyMqConfig中添加：
@Bean
public Topic topic() {
return new ActiveMQTopic("pic");
stc2052}
这⾥创建了ActiveMQTopic，并将topic的名称指定为pic。
Producer中新增如下代码：
@Resource
private Topic topic;
public void sendTopic(String msg) {
System.out.println("发送Topic消息内容 :"+msg);
pic, msg);
}
为了演⽰多个⼴播接收者，在Comsumer中新增两个消费者：
@JmsListener(destination = "pic")
public void receiveTopic1(String text) {
System.out.println("receiveTopic1接收到Topic消息 : " + text);
}
@JmsListener(destination = "pic")
public void receiveTopic2(String text) {
System.out.println("receiveTopic2接收到Topic消息 : " + text);
}
单元测试类中新增如下测试：
@Test
public void sendSimpleTopicMessage() {
this.producer.sendTopic("提现200.00元");
}
此时，执⾏单元测试，便可看到如下⽇志信息：
发送Topic消息内容 ：提现200.00元
receiveTopic2接收到Topic消息 : 提现200.00元
receiveTopic1接收到Topic消息 : 提现200.00元
说明消息发送成功。
深度时空同时⽀持两种形式
在上⾯的实例中，要么⽀持队列模式要么⽀持⼴播模式，如果在⽣产环境中两者都需要⽀持，那么就需要⾃定义JmsListenerContainerFactory实例。当然，如果Spring Boot默认的配置⽆法满⾜需求，也可以⾃定义该类，这⾥只是其中场景之⼀。
基本配置和使⽤步骤：通过DefaultJmsListenerContainerFactory创建⾃定义的JmsListenerContainerFactory实例，在
@JmsListener注解中通过containerFactory属性进⾏引⽤。
在MyMqConfig配置类中新增如下配置：
@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属性。⽰例代码如下：@JmsListener(destination = "sms.queue", containerFactory = "queueListenerFactory")
public void receiveMsg(String text) {
System.out.println("接收到消息 : " + text);
}
@JmsListener(destination = "pic", containerFactory = "topicListenerFactory")
public void receiveTopic1(String text) {
System.out.println("receiveTopic1接收到Topic消息 : " + text);
}
分别执⾏两种形式的消息，发现都正常互利。同时，此时配置⽂件中的项spring.jms.pub-sub-domain也⽆效了。
其他事项
1、activeMq的端⼝号是61616；
2、使⽤topic，需要配置spring.jms.pub-sub-domain=true;
3、queue如果没有消费者，会将信息存储到queue中；
4、发送的消息为对象的时候，需要将对象序列化；消费者接收对象信息时需要使⽤ObjectMessage进⾏转化；
5、使⽤JmsListener注解中的containerFactory属性，可以配置spring.jms.pub-sub属性，实现同时接
收queque和topic；
6、queue为点对点模式；tipic为发布订阅模式；
7、⽰例中的消息队列名称（sms.queue和pic）可根据需要设置成配置属性；
源码地址：…
参考⽂章：
总结
到此这篇关于SpringBoot集成ActiveMQ的⽂章就介绍到这了，更多相关SpringBoot集成ActiveMQ内容请搜索以前的⽂章或继续浏览下⾯的相关⽂章希望⼤家以后多多⽀持！
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议