ActiveMQ 笔记（四）Spring\SpringBoot 整合 Activemq
个人博客网：https://wushaopei.github.io/ (你想要这里多有)
一、Spring 整合Activemq
1、所需jar包
org.apache.activemq
activemq-all
5.15.11
org.apache.activemq
activemq-pool
5.15.10
org.apache.xbean
xbean-spring
4.15
com.fasterxml.jackson.core
jackson-databind
2.10.1
org.springframework
spring-jms
5.2.1.RELEASE
org.springframework
spring-aop
5.2.1.RELEASE
2、Spring配置文件（applicationContext.xml）
<?xml version="1.0" encoding="UTF-8"?>
3、编写代码
（1）队列（Queue）
@Service
public class SpringMQ_Producer {
private JmsTemplate jmsTemplate;
@Autowired
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml");
SpringMQ_Producer springMQ_producer = applicationContext.getBean(SpringMQ_Producer.class);
springMQ_producer.jmsTemplate.send(session -> session.createTextMessage("***Spring和ActiveMQ的整合case111....."));
System.out.println("********send task over");
}
}
@Service
public class SpringMQ_Consumer {
private JmsTemplate jmsTemplate;
@Autowired
public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml");
SpringMQ_Consumer springMQ_consumer = applicationContext.getBean(SpringMQ_Consumer.class);
String returnValue = (String) springMQ_consumer.jmsTemplate.receiveAndConvert();
System.out.println("****消费者收到的消息: " + returnValue);
}
}
（2）主题（Topic）
生产者和消费者都可以通过jmsTemplate对象实时设置目的地等其他信息
@Service
public class SpringMQ_Topic_Producer {
private JmsTemplate jmsTemplate;
public SpringMQ_Topic_Producer(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml");
SpringMQ_Topic_Producer springMQ_topic_producer = applicationContext.getBean(SpringMQ_Topic_Producer.class);
//直接调用application.xml里面创建的destinationTopic这个bean设置为目的地就行了
springMQ_topic_producer.jmsTemplate.setDefaultDestination(((Destination) applicationContext.getBean("destinationTopic")));
springMQ_topic_producer.jmsTemplate.send(session -> session.createTextMessage("***Spring和ActiveMQ的整合TopicCase111....."));
}
}@Service
public class SpringMQ_Topic_Consumer {
private JmsTemplate jmsTemplate;
public SpringMQ_Topic_Consumer(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public static void main(String[] args) {
ApplicationContext applicationContext = new ClassPathXmlApplicationContext("Application.xml");
SpringMQ_Topic_Consumer springMQConsumer = applicationContext.getBean(SpringMQ_Topic_Consumer.class);
//直接调用application.xml里面创建的destinationTopic这个bean设置为目的地就行了
springMQConsumer.jmsTemplate.setDefaultDestination(((Destination) applicationContext.getBean("destinationTopic")));
String returnValue = (String) springMQConsumer.jmsTemplate.receiveAndConvert();
System.out.println("****消费者收到的消息: " + returnValue);
}
}
4、监听配置
在Spring里面实现消费者不启动，直接通过配置监听完成（1）说明
类似于前面setMessageListenner实时间提供消息（2）Spring 配置文件
<?xml version="1.0" encoding="UTF-8"?>
（3）需要写一个类来实现消息监听
/**
* 实现MessageListener的类,需要把这个类交给xml配置里面的DefaultMessageListenerContainer管理
*/
@Component
public class MyMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("消费者收到的消息" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
（4）消费者配置了自动监听，就相当于在spring里面后台运行，有消息就运行我们实现监听类里面的方法
二、SpringBoot 整合Activemq
1、队列（Queue）
1.1 队列生产者：
（1）新建Maven工程并设置包名类名
工程名： boot_mq_producer
包名： com.atguigu.boot.activemq（2）POM依赖文件配置：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0modelVersion>
<groupId>com.demogroupId>
<artifactId>04activemq_springboot-queueartifactId>
<version>1.0-SNAPSHOTversion>
<dependencies>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starterartifactId>
<version>2.2.1.RELEASEversion>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-activemqartifactId>
<version>2.2.1.RELEASEversion>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-webartifactId>
<version>2.2.1.RELEASEversion>
dependency>
<dependency>
<groupId>org.springframework.bootgroupId>
<artifactId>spring-boot-starter-testartifactId>
<version>2.2.1.RELEASEversion>
dependency>
dependencies>
project>
（3）application.yml 文件
#Springboot启动端口
server:
port: 8080
#ActiveMQ配置
spring:
activemq:
broker-url: tcp://192.168.10.130:61616 #ActiveMQ服务器IP
user: admin #ActiveMQ连接用户名
password: admin #ActiveMQ连接密码
jms:
#指定连接队列还是主题
pub-sub-domain: false # false = Queue | true = Topic
#定义服务上的队列名
myQueueName: springboot-activemq-queue（4）配置队列bean
类似于Spring的ApplicationContext.xml文件
@Component
@EnableJms //开启Springboot的Jms
public class ConfigBean {
@Value("myQueueName")
private String myQueueName;
@Bean
public ActiveMQQueue queue() {
//创建一个ActiveMQQueue
return new ActiveMQQueue(myQueueName);
}
}（5）队列生产者
@Component
public class Queue_Producer {
//JmsMessagingTemplate是Springboot的Jms模板,Spring的是JmsTemplate
private JmsMessagingTemplate jmsMessagingTemplate;
//把ConfigBean类的ActiveMQQueue注入进来
private ActiveMQQueue activeMQQueue;
//发送Queue的方法
public void producerMsg() {
jmsMessagingTemplate.convertAndSend(activeMQQueue, "**************" + UUID.randomUUID().toString());
}
//构造注入对象(推荐)
public Queue_Producer(JmsMessagingTemplate jmsMessagingTemplate, ActiveMQQueue activeMQQueue) {
this.jmsMessagingTemplate = jmsMessagingTemplate;
this.activeMQQueue = activeMQQueue;
}
}
（6）测试单元
@SpringBootTest(classes = MainApp.class)
@RunWith(SpringJUnit4ClassRunner.class)
@WebAppConfiguration
public class TestActiveMQ {
@Autowired
private Queue_Producer queue_producer;
@Test
public void testSend() {
queue_producer.producerMsg();
}
}
结果： 测试结果应该是消息正常生产，并在控制台能够看到入列数大于0且为消费消息大于0
1.2 队列消费者
（1）1.新建Mavaen工程并设置包名类名
（2）Pom依赖配置与生产者一致
（3）application.yml
server:
port: 8081
这里仅对端口进行修改，其他配置与生产者一致
（4）消息消费者
@Component
public class Queue_consummer {
@JmsListener(destination = "${myqueue}") // 注解监听
public void receive(TextMessage textMessage) throws Exception{
System.out.println(" *** 消费者收到消息 ***"+textMessage.getText());
}
}1.3 定时投递与监听消费
新需求: 要求每隔3秒钟,往MQ推送消息 以下定时发送Case，案例修改分析： 在需要定时推送消息的情况下，同时也需要对消息进行跟踪消费，也就是监听消息的生产并进行消费。
生产者代码：
（1）修改Queue_Produce新增定时投递方法
@Component
public class Queue_Producer {
//间隔3秒投递,SpringBoot的Scheduled用来定时执行
@Scheduled(fixedDelay = 3000)
public void producerMsgScheduled() {
jmsMessagingTemplate.convertAndSend(activeMQQueue, "**************Scheduled" + UUID.randomUUID().toString());
System.out.println("Scheduled定时投递");
}
（2）修改主启动类的MainApp_Producer
@SpringBootApplication
@EnableScheduling
public class MainApp {
public static void main(String[] args) {
SpringApplication.run(MainApp.class);
}
}在主启动类这里主要是要添加@EnableScheduling 注解，用来开启定时配置
消息消费者代码增强
（1）springboot的消息监听注解
@JmsListener说明：监听过后会随着springboot一起启动,有消息就执行加了该注解的方法
（2）具体实现代码：
//监听接收的方法
@JmsListener(destination = "${myQueueName}")
public void consumerMsg(TextMessage textMessage) throws JMSException {
String text = textMessage.getText();
System.out.println("***消费者收到的消息: " + text);
}（3）主启动类
@SpringBootApplication
@EnableScheduling // 开启消息定投功能
public class MainApp_Produce {
public static void main(String[] args) {
SpringApplication.run(MainApp_Produce.class, args);
}
}2、主题发布订阅（Topic）
2.1 Topic 生产者
（1）新建Maven工程并设置包名类名
（2）POM文件
订阅（Topic）的依赖于队列（Queue）的一致
（3）Yml文件
jms:
pub-sub-domain: true
myTopicName: springboot-activemq-topic（4）配置bean
@Component
@EnableJms //开启Springboot的Jms
public class ActiveMQConfigBean {
@Value("${myTopicName}")
private String topicName;
@Bean
public ActiveMQTopic activeMQTopic() {
return new ActiveMQTopic(topicName);
}
}注意：类似于Spring的ApplicationContext.xml文件
（5）Topic生产者代码：
@Component
@EnableScheduling
public class Topic_Producer {
private JmsMessagingTemplate jmsMessagingTemplate;
private ActiveMQTopic activeMQTopic;
@Scheduled(fixedDelay = 3000)
public void producer() {
jmsMessagingTemplate.convertAndSend(activeMQTopic, "主题消息: " + UUID.randomUUID().toString());
}2.2 Topic 消费者
（1）新建Maven工程并设置包名类名
（2）Pom （同上）
（3）yml文件
jms:
pub-sub-domain: true
myTopicName: springboot-activemq-topic（4）配置bean
/**
* 设置持久化订阅
* 配置文件的方式无法进行配置持久化订阅。所以需要自己去生成一个持久化订阅
*/
@Component
@EnableJms
public class ActiveMQConfigBean {
@Value("${spring.activemq.broker-url}")
private String brokerUrl;
@Value("${spring.activemq.user}")
private String user;
@Value("${spring.activemq.password}")
private String password;
public ConnectionFactory connectionFactory(){
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(brokerUrl);
connectionFactory.setUserName(user);
connectionFactory.setPassword(password);
return connectionFactory;
}
@Bean(name = "jmsListenerContainerFactory")
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
DefaultJmsListenerContainerFactory defaultJmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
defaultJmsListenerContainerFactory.setConnectionFactory(connectionFactory());
defaultJmsListenerContainerFactory.setSubscriptionDurable(true);
defaultJmsListenerContainerFactory.setClientId("我是持久订阅者一号");
return defaultJmsListenerContainerFactory;
}
配置文件的方式无法进行配置持久化订阅。所以需要自己去生成一个持久化订阅
（5）Topic消费者代码
@Component
public class Topic_Consumer {
//需要在监听方法指定连接工厂
@JmsListener(destination = "${myTopicName}",containerFactory = "jmsListenerContainerFactory")
public void consumer(TextMessage textMessage) throws JMSException {
System.out.println("订阅着收到消息: " + textMessage.getText());
}
}重点：
@JmsListener(destination = "${myTopicName}",containerFactory = "jmsListenerContainerFactory")启动顺序：先启动消费者,后启动生产者