ActiveMQ 6. Spring整合ActiveMQ
6.1. Spring对ActiveMQ队列、主题的整合
- pom.xml
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<dependency>
<groupId>org.apache.xbean</groupId>
<artifactId>xbean-spring</artifactId>
<version>3.16</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.9.5</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.10</version>
</dependency>
<!-- Spring对JMS的支持 -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
<version>5.2.3.RELEASE</version>
</dependency>
<!--ActiveMQ所需要的pool包配置-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
<version>5.15.9</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-aop</artifactId>
<version>5.2.3.RELEASE</version>
</dependency>
-
Spring配置文件
这里在JmsTemplate的defaultDestination属性上配置的是队列,如果需要使用主题,只需要修改这个属性值,其他代码不用修改。
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd">
<!-- 开启包的自动扫描 -->
<context:component-scan base-package="study.hwj.activemq"/>
<!-- 配置生产者 -->
<bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
<property name="connectionFactory">
<!-- 真正可以生产Connection的ConnectionFactory,由对应的JMS服务商提供 -->
<bean class="org.apache.activemq.spring.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.181.128:61616"/>
</bean>
</property>
<property name="maxConnections" value="100"/>
</bean>
<!-- 这个是队列目的地,点对点的Queue -->
<bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 通过构造注入Queue名 -->
<constructor-arg index="0" value="spring-activemq-queue"/>
</bean>
<!-- 这个是队列目的地, 发布订阅的主题Topic-->
<bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg index="0" value="spring-activemq-topic"/>
</bean>
<!-- Spring提供的JMS工具类,他可以进行消息发送,接收等 -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<!-- 传入连接工厂 -->
<property name="connectionFactory" ref="connectionFactory"/>
<!-- 传入目的地 -->
<property name="defaultDestination" ref="destinationQueue"/>
<!-- 消息自动转换器 -->
<property name="messageConverter">
<bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
</property>
</bean>
</beans>
-
生产者代码
@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("applicationContext.xml"); SpringMQ_Producer springMQ_producer = applicationContext.getBean(SpringMQ_Producer.class); springMQ_producer.jmsTemplate.send(new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage("***Spring和ActiveMQ的整合case111....."); } }); // 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("applicationContext.xml");
SpringMQ_Consumer springMQ_consumer = applicationContext.getBean(SpringMQ_Consumer.class);
String returnValue = (String) springMQ_consumer.jmsTemplate.receiveAndConvert();
System.out.println("****消费者收到的消息: " + returnValue);
}
}
6.2. ActiveMQ主题监听器
-
Spring配置文件
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd"> <!-- 开启包的自动扫描 --> <context:component-scan base-package="study.hwj.activemq"/> <!-- 配置生产者 --> <bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory"> <!-- 真正可以生产Connection的ConnectionFactory,由对应的JMS服务商提供 --> <bean class="org.apache.activemq.spring.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://192.168.181.128:61616"/> </bean> </property> <property name="maxConnections" value="100"/> </bean> <!-- 这个是队列目的地,点对点的Queue --> <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue"> <!-- 通过构造注入Queue名 --> <constructor-arg index="0" value="spring-activemq-queue"/> </bean> <!-- 这个是队列目的地, 发布订阅的主题Topic--> <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg index="0" value="spring-activemq-topic"/> </bean> <!-- Spring提供的JMS工具类,他可以进行消息发送,接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 传入连接工厂 --> <property name="connectionFactory" ref="connectionFactory"/> <!-- 传入目的地 --> <property name="defaultDestination" ref="destinationTopic"/> <!-- 消息自动转换器 --> <property name="messageConverter"> <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/> </property> </bean> <!-- 配置Jms消息监听器 --> <bean id="defaultMessageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <!-- Jms连接的工厂 --> <property name="connectionFactory" ref="connectionFactory"/> <!-- 设置默认的监听目的地 --> <property name="destination" ref="destinationTopic"/> <!-- 指定自己实现了MessageListener的类 --> <property name="messageListener" ref="myMessageListener"/> </bean> <bean id="myMessageListener" class="study.hwj.activemq.MyMessageListener"/> </beans>
-
监听器代码
/**
* 实现MessageListener的类,需要把这个类交给xml配置里面的DefaultMessageListenerContainer管理
*/
public class MyMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("Listener收到的消息" + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
-
生产者代码
与6.1. 中的生产者代码相同