Spring與ActiveMQ整合
阿新 • • 發佈：2019-01-26
maven依賴：
activemq.properties內容：<!-- https://mvnrepository.com/artifact/org.springframework/spring-jms --> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>4.3.2.RELEASE</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.xbean/xbean-spring --> <dependency> <groupId>org.apache.xbean</groupId> <artifactId>xbean-spring</artifactId> <version>4.5</version> </dependency> <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-core --> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-core</artifactId> <version>5.7.0</version> </dependency>
brokerURL=tcp://127.0.0.1:61616
activemq.userName=admin
activemq.password=admin
queue.address=queue
topic.address=topic訊息生產者檔案的內容：<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:jms="http://www.springframework.org/schema/jms" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.2.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-4.2.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.7.0.xsd"> <!-- 引入配置檔案 --> <bean id="activemqPropertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"> <property name="order" value="3" /> <property name="ignoreUnresolvablePlaceholders" value="true" /> <property name="locations"> <list> <value>classpath:activemq.properties</value> </list> </property> </bean> <!-- ActiveMQ 連線工廠 --> <!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供--> <!-- <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin" /> --> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="${brokerURL}"/> <property name="userName" value="${activemq.userName}"/> <property name="password" value="${activemq.password}"/> </bean> <!-- Spring 連線工廠 --> <!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory --> <!-- Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。 SingleConnectionFactory對於建立JMS伺服器連結的請求會一直返回同一個連結，並且會忽略Connection的close方法呼叫。 CachingConnectionFactory繼承了SingleConnectionFactory，所以它擁有SingleConnectionFactory的所有功能， 同時它還新增了快取功能，它可以快取Session、MessageProducer和MessageConsumer。--> <bean id="jmsconnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory"/> <!-- Session快取數量 --> <property name="sessionCacheSize" value="20" /> </bean> <!-- 定義訊息的目的地 --> <!--點對點訊息佇列目的地--> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg value="${queue.address}"/> </bean> <!--釋出訂閱主題目的地--> <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg value="${topic.address}"/> </bean> <!-- 訊息生產者 start --> <!-- 定義JmsTemplate的Queue型別 --> <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 --> <property name="connectionFactory" ref="jmsconnectionFactory"/> <property name="defaultDestination" ref="queueDestination" /> <!-- 如果不用defaultDestination而是用defaultDestinationName則不需要在前面配置目的地了 --> <!-- <property name="defaultDestinationName" value="${queue.address}"/> --> <!-- 非pub/sub模型（釋出/訂閱），即佇列模式 --> <property name="pubSubDomain" value="false" /> </bean> <!-- 定義JmsTemplate的Topic型別 --> <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 --> <property name="connectionFactory" ref="jmsconnectionFactory"/> <property name="defaultDestination" ref="topicDestination" /> <!-- 如果不用defaultDestination而是用defaultDestinationName則不需要在前面配置目的地了 --> <!-- <property name="defaultDestinationName" value="${topic.address}"/> --> <!-- pub/sub模型（釋出/訂閱） --> <property name="pubSubDomain" value="true" /> </bean> <!--訊息生產者 end--> <!-- 訊息消費者 start--> <!-- 定義Queue監聽器 --> <jms:listener-container destination-type="queue" container-type="default" connection-factory="jmsconnectionFactory" acknowledge="auto"> <!-- 在這裡可以有很多的訊息消費者，ref後是spring掃描檔案時注入的bean --> <jms:listener destination="${queue.address}" ref="queueConsumer1"/> <jms:listener destination="${queue.address}" ref="queueConsumer2"/> </jms:listener-container> <!-- 定義Topic監聽器 --> <jms:listener-container destination-type="topic" container-type="default" connection-factory="jmsconnectionFactory" acknowledge="auto"> <!-- 在這裡可以有很多的訊息消費者，ref後是spring掃描檔案時注入的bean --> <jms:listener destination="${topic.address}" ref="topicConsumer1"/> <jms:listener destination="${topic.address}" ref="topicConsumer2"/> </jms:listener-container> <!-- 訊息消費者 end --> </beans>
QueueProducer.java檔案：
TopicProducer.java檔案：package com.cn.common.jmsProducer; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; import org.springframework.stereotype.Component; @Component public class QueueProducer { @Autowired @Qualifier("jmsQueueTemplate") private JmsTemplate jmsTemplate; //注入配置檔案中的bean @Autowired @Qualifier("queueDestination") private Destination destination; public void send(final String message){ /*jmsTemplate.send(destination, new MessageCreator() { @Override public Message createMessage(Session paramSession) throws JMSException { return paramSession.createTextMessage(message); } });*/ jmsTemplate.send(new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } });; } }
package com.cn.common.jmsProducer;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;
@Component
public class TopicProducer {
@Autowired
@Qualifier("jmsTopicTemplate")
private JmsTemplate jmsTemplate; //注入配置檔案中的bean
@Autowired
@Qualifier("topicDestination")
private Destination destination;
public void send(final String message){
/*jmsTemplate.send(destination, new MessageCreator() {
@Override
public Message createMessage(Session paramSession) throws JMSException {
return paramSession.createTextMessage(message);
}
});*/
jmsTemplate.send(new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(message);
}
});;
}
}
QueueConsumer1.java檔案：
package com.cn.common.jmsConsumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.springframework.stereotype.Component;
@Component
public class QueueConsumer1 implements MessageListener{
@Override
public void onMessage(Message message) {
try {
TextMessage text = null;
if(message instanceof TextMessage){
text = (TextMessage) message;
System.out.println("QueueConsumer1接收到訊息："+text.getText());
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
package com.cn.common.jmsConsumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.springframework.stereotype.Component;
@Component
public class QueueConsumer2 implements MessageListener{
@Override
public void onMessage(Message message) {
try {
TextMessage text = null;
if(message instanceof TextMessage){
text = (TextMessage) message;
System.out.println("QueueConsumer2接收到訊息："+text.getText());
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
package com.cn.common.jmsConsumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.springframework.stereotype.Component;
@Component
public class TopicConsumer1 implements MessageListener{
@Override
public void onMessage(Message message) {
try {
TextMessage text = null;
if(message instanceof TextMessage){
text = (TextMessage) message;
System.out.println("TopicConsumer1接收到訊息："+text.getText());
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
package com.cn.common.jmsConsumer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.springframework.stereotype.Component;
@Component
public class TopicConsumer2 implements MessageListener{
@Override
public void onMessage(Message message) {
try {
TextMessage text = null;
if(message instanceof TextMessage){
text = (TextMessage) message;
System.out.println("TopicConsumer2接收到訊息："+text.getText());
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}