spring整合使用activemq

阿新 • • 發佈：2018-12-12

activemq簡單的demo這裡就不做演示了，今天介紹一下如何利用spring整合activemq，也是實際工作中涉及到的，希望對各位夥伴有所幫助，

1、安裝activemq，為演示方便，我已經提前在linux中安裝了activemq，安裝過程比較簡單，應該說activemq是軟體中算是最簡單的一種了，只需要解壓，進入bin目錄啟動即可，不需做其他的配置；在這裡插入圖片描述

只需要一條命令，即可啟動，注意需要把activemq的埠配置到防護牆的檔案中，然後重啟防火牆，否則瀏覽器直接訪問是訪問不到的，

在這個頁面可以進行相關的配置，具體可參考官網說明，為後面的使用，我這裡提前建立一個佇列名稱，test_queue,後面可以直接拿來使用；

接下來是專案和程式碼的部分；

2、專案結構，在這裡插入圖片描述

為模擬真實的場景，這裡我建立了兩個maven專案，首先看producer的部分，

在這裡插入圖片描述 pom.xml檔案，

<dependencies>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-core</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-context</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-orm</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-web</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jdbc</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-webmvc</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<!-- spring aop -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-aspects</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>com.mchange</groupId>
			<artifactId>c3p0</artifactId>
			<version>0.9.2</version>
		</dependency>
		<dependency>
			<groupId>cglib</groupId>
			<artifactId>cglib</artifactId>
			<version>2.2</version>
		</dependency>
		<dependency>
			<groupId>org.aspectj</groupId>
			<artifactId>aspectjweaver</artifactId>
			<version>1.6.8</version>
		</dependency>

		<!-- Spring整合MyBatis -->
		<dependency>
			<groupId>org.mybatis</groupId>
			<artifactId>mybatis</artifactId>
			<version>3.2.8</version>
		</dependency>

		<dependency>
			<groupId>org.mybatis</groupId>
			<artifactId>mybatis-spring</artifactId>
			<version>1.2.2</version>
		</dependency>

		<!-- 控制日誌輸出：結合log4j -->
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.17</version>
		</dependency>

		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
			<version>1.7.7</version>
		</dependency>

		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>5.1.37</version>
		</dependency>

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-context-support</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<dependency>
			<groupId>commons-collections</groupId>
			<artifactId>commons-collections</artifactId>
			<version>3.1</version>
		</dependency>

		<dependency>
			<groupId>commons-codec</groupId>
			<artifactId>commons-codec</artifactId>
			<version>1.9</version>
		</dependency>

		<dependency>
			<groupId>commons-io</groupId>
			<artifactId>commons-io</artifactId>
			<version>2.4</version>
		</dependency>

		<dependency>
			<groupId>org.codehaus.jackson</groupId>
			<artifactId>jackson-mapper-asl</artifactId>
			<version>1.9.2</version>
		</dependency>

		<dependency>
			<groupId>commons-fileupload</groupId>
			<artifactId>commons-fileupload</artifactId>
			<version>1.3.1</version>
		</dependency>

		<dependency>
			<groupId>javax.servlet</groupId>
			<artifactId>servlet-api</artifactId>
			<version>2.5</version>
		</dependency>

		<dependency>
			<groupId>javax.servlet.jsp</groupId>
			<artifactId>jsp-api</artifactId>
			<version>2.1.3-b06</version>
		</dependency>

		<dependency>
			<groupId>jstl</groupId>
			<artifactId>jstl</artifactId>
			<version>1.2</version>
		</dependency>

		<!-- spring單元測試 -->
		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-test</artifactId>
			<version>4.3.7.RELEASE</version>
			<scope>test</scope>
		</dependency>

		<!-- 格式化物件，方便輸出日誌 -->
		<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>fastjson</artifactId>
			<version>1.1.41</version>
		</dependency>

		<dependency>
			<groupId>com.alibaba</groupId>
			<artifactId>druid</artifactId>
			<version>1.0.9</version>
		</dependency>

		<dependency>
			<groupId>org.aspectj</groupId>
			<artifactId>aspectjweaver</artifactId>
			<version>1.8.8</version>
		</dependency>

		<!--rabbitmq依賴 -->
		<!-- <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> 
			<version>3.6.0</version> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> 
			<artifactId>spring-rabbit</artifactId> <version>1.6.5.RELEASE</version> </dependency> -->

		<dependency>
			<groupId>org.springframework</groupId>
			<artifactId>spring-jms</artifactId>
			<version>4.0.0.RELEASE</version>
		</dependency>

		<!--active mq start -->
		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-core</artifactId>
			<version>5.7.0</version>
		</dependency>

		<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-pool</artifactId>
			<version>5.13.3</version>
		</dependency>
		<!--active mq end -->
	</dependencies>

大家可以根據自己需要，把主要的配置檔案引入即可，整合spring的時候，倒數第三個是必須的；

log4j.properties老套路，這裡不多解釋，請看mq.properties，這個裡面是連線linux中的activemq的主要配置資訊，

## MQ
mq.brokerURL=tcp\://192.168.111.130\:61616
mq.userName=admin
mq.password=admin
mq.pool.maxConnections=10
#queueName
queueName=test_queue

spring-mq.xml，這個配置檔案裡，配置的項如ConnectionFactory，池化連線，activeMqJmsTemplate（佇列模板）等，主要是方便專案程式碼中的連線使用，方便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:p="http://www.springframework.org/schema/p"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans  
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
           http://www.springframework.org/schema/aop   
           http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
           http://www.springframework.org/schema/tx  
           http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
           http://www.springframework.org/schema/context  
           http://www.springframework.org/schema/context/spring-context-3.2.xsd"
	default-autowire="byName" default-lazy-init="false">
	
	<!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<!-- ActiveMQ服務地址 -->
        <property name="brokerURL" value="${mq.brokerURL}" />
        <property name="userName" value="${mq.userName}"></property>
        <property name="password" value="${mq.password}"></property> 
	</bean>

    <!-- 
    	ActiveMQ為我們提供了一個PooledConnectionFactory，通過往裡面注入一個ActiveMQConnectionFactory
    	可以用來將Connection、Session和MessageProducer池化，這樣可以大大的減少我們的資源消耗。
    	要依賴於 activemq-pool包
     -->
	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
		<property name="connectionFactory" ref="targetConnectionFactory" />
		<property name="maxConnections" value="${mq.pool.maxConnections}" />
	</bean>

	<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
		<!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
	</bean>
	
	<!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 -->
	<!-- 佇列模板 -->
	<bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
	    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 -->  
	    <property name="connectionFactory" ref="connectionFactory"/>  
	    <property name="defaultDestinationName" value="${queueName}"></property>
	</bean> 

</beans>

接下來是和spring配置檔案整合在一起，請看spring-context.xml配置檔案，

<?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:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans  
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
           http://www.springframework.org/schema/aop   
           http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
           http://www.springframework.org/schema/tx  
           http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
           http://www.springframework.org/schema/context  
           http://www.springframework.org/schema/context/spring-context-3.2.xsd"
	default-autowire="byName" default-lazy-init="false">
	
	<!-- 採用註釋的方式配置bean -->
	<context:annotation-config />

	<!-- 配置要掃描的包 -->
	<context:component-scan base-package="wusc.edu.demo" />

	<!-- 讀入配置屬性檔案 -->
	<context:property-placeholder location="classpath:mq.properties" />

	<!-- proxy-target-class預設"false",更改為"ture"使用CGLib動態代理 -->
	<aop:aspectj-autoproxy proxy-target-class="true" />	
	
	<import resource="spring-mq.xml" />
	
</beans>

裡面主要包括，引入mq的配置檔案，掃描包，載入spring-mq的配置檔案，比較簡單，相信大家一眼就能看明白；

下面是具體的程式碼，這裡為了模擬生產者，使用簡單的點對點的方式進行訊息的傳送； MailParam是個實體類，可以作為一個物件進行傳輸，

public class MailParam {
	
	/** 發件人 **/
	private String from;
	/** 收件人 **/
	private String to;
	/** 主題 **/
	private String subject;
	/** 郵件內容 **/
	private String content;

	public MailParam() {
	}

	public MailParam(String to, String subject, String content) {
		this.to = to;
		this.subject = subject;
		this.content = content;
	}

	public String getFrom() {
		return from;
	}

	public void setFrom(String from) {
		this.from = from;
	}

	public String getTo() {
		return to;
	}

	public void setTo(String to) {
		this.to = to;
	}

	public String getSubject() {
		return subject;
	}

	public void setSubject(String subject) {
		this.subject = subject;
	}

	public String getContent() {
		return content;
	}

	public void setContent(String content) {
		this.content = content;
	}
}

MQProducer：

@Service("mqProducer")
public class MQProducer {
	
	@Autowired
	private JmsTemplate activeMqJmsTemplate;
	
	public void sendMessage(final MailParam mail){
		activeMqJmsTemplate.send(new MessageCreator() {
			
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(JSONObject.toJSONString(mail));
			}
		});
	}
	
}

在test目錄下建立一個測試類MQProducerTest，

public class MQProducerTest {
	
	private static final Log log = LogFactory.getLog(MQProducerTest.class);
	
	public static void main(String[] args) {
		try {
			
			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");
			context.start();
			
			MQProducer mqProducer = (MQProducer) context.getBean("mqProducer");
			
			MailParam mail = new MailParam();
			mail.setTo("[email protected]");
			mail.setSubject("ActiveMQ測試");
			mail.setContent("通過ActiveMQ非同步傳送郵件！");
			
			mqProducer.sendMessage(mail);
			
		} catch (Exception e) {
			log.error("==>MQ context start error:", e);
			System.exit(0);
		}finally{
			log.info("===>System.exit");
			System.exit(0);
		}
	}
	
	
}

再看消費端，在這裡插入圖片描述

pom檔案和producer基本一致，不再羅列，配置檔案部門也差不多，從專案結構可以看出來，在和spring進行整合的時候，消費端常常需要設定一個監聽器，進行非同步的監聽消費，這個和rocketmq或rabbitmq思想類似，log4j.properties不再羅列，mq.properties如下，

## MQ
mq.brokerURL=tcp\://192.168.111.130\:61616
mq.userName=admin
mq.password=admin
mq.pool.maxConnections=10
#queueName
queueName=test_queue

注意到，生產者個消費端都採用了連線池，用於管理併發的連線數，方便管理和控制，這一點請注意；

spring-mq.xml，這個和producer中類似，多加了訊息監聽器的配置部分，

<?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:p="http://www.springframework.org/schema/p"
	xmlns:context="http://www.springframework.org/schema/context"
	xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans  
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
           http://www.springframework.org/schema/aop   
           http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
           http://www.springframework.org/schema/tx  
           http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
           http://www.springframework.org/schema/context  
           http://www.springframework.org/schema/context/spring-context-3.2.xsd"
	default-autowire="byName" default-lazy-init="false">

	<!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 -->
	<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<!-- ActiveMQ服務地址 -->
        <property name="brokerURL" value="${mq.brokerURL}" />
        <property name="userName" value="${mq.userName}"></property>
        <property name="password" value="${mq.password}"></property> 
	</bean>

    <!-- 
    	ActiveMQ為我們提供了一個PooledConnectionFactory，通過往裡面注入一個ActiveMQConnectionFactory
    	可以用來將Connection、Session和MessageProducer池化，這樣可以大大的減少我們的資源消耗。
    	要依賴於 activemq-pool包
     -->
	<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
		<property name="connectionFactory" ref="targetConnectionFactory" />
		<property name="maxConnections" value="${mq.pool.maxConnections}" />
	</bean>

	<!-- Spring用於管理真正的ConnectionFactory的ConnectionFactory -->
	<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
		<!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory -->
		<property name="targetConnectionFactory" ref="pooledConnectionFactory" />
	</bean>
	
	<!-- Spring提供的JMS工具類，它可以進行訊息傳送、接收等 -->
	
	<!-- 佇列模板 -->
	<bean id="activeMqJmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
	    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory物件 -->  
	    <property name="connectionFactory" ref="connectionFactory"/>  
	    <property name="defaultDestinationName" value="${queueName}"></property>
	</bean> 
	
	<!--這個是sessionAwareQueue目的地 -->
	<bean id="sessionAwareQueue" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg>
			<value>${queueName}</value>
		</constructor-arg>
	</bean>

	<!-- 可以獲取session的MessageListener -->
	<bean id="consumerSessionAwareMessageListener" class="wusc.edu.demo.mqtest.listener.ConsumerSessionAwareMessageListener"></bean>

	<bean id="sessionAwareListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<property name="connectionFactory" ref="connectionFactory" />
		<property name="destination" ref="sessionAwareQueue" />
		<property name="messageListener" ref="consumerSessionAwareMessageListener" />
	</bean>

</beans>

spring-context.xml，用於引入外部配置檔案並整合activemq的配置檔案，

<?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:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
	xmlns:tx="http://www.springframework.org/schema/tx"
	xsi:schemaLocation="http://www.springframework.org/schema/beans  
           http://www.springframework.org/schema/beans/spring-beans-3.2.xsd  
           http://www.springframework.org/schema/aop   
           http://www.springframework.org/schema/aop/spring-aop-3.2.xsd  
           http://www.springframework.org/schema/tx  
           http://www.springframework.org/schema/tx/spring-tx-3.2.xsd  
           http://www.springframework.org/schema/context  
           http://www.springframework.org/schema/context/spring-context-3.2.xsd"
	default-autowire="byName" default-lazy-init="false">
	
	<!-- 採用註釋的方式配置bean -->
	<context:annotation-config />

	<!-- 配置要掃描的包 -->
	<context:component-scan base-package="wusc.edu.demo" />

	<!-- 讀入配置屬性檔案 -->
	<context:property-placeholder location="classpath:mq.properties,classpath:mail.properties" />

	<!-- proxy-target-class預設"false",更改為"ture"使用CGLib動態代理 -->
	<aop:aspectj-autoproxy proxy-target-class="true" />	
	
	<import resource="spring-mq.xml" />
	<import resource="spring-mail.xml" />
	
</beans>

接下來，請看訊息監聽部分的程式碼，ConsumerSessionAwareMessageListener，這個類需要實現一個介面，SessionAwareMessageListener，

@Component
public class ConsumerSessionAwareMessageListener implements SessionAwareMessageListener<Message> {

	private static final Log log = LogFactory.getLog(ConsumerSessionAwareMessageListener.class);

	@Autowired
	private JmsTemplate activeMqJmsTemplate;
	
	@Autowired
	private Destination sessionAwareQueue;
	
	@Autowired
	private MailBiz bailBiz;
	
	public synchronized void onMessage(Message message, Session session) {
		try {
			ActiveMQTextMessage msg = (ActiveMQTextMessage) message;
			final String ms = msg.getText();
			log.info("==>receive message:" + ms);
			MailParam mailParam = JSONObject.parseObject(ms, MailParam.class);// 轉換成相應的物件
			if (mailParam == null) {
				return;
			}

			try {
				System.out.println("收到的訊息內容是==============>>>>>>>>>>>"+mailParam.getContent());
				//bailBiz.mailSend(mailParam);
			} catch (Exception e) {
				// 傳送異常，重新放回佇列
//				activeMqJmsTemplate.send(sessionAwareQueue, new MessageCreator() {
//					public Message createMessage(Session session) throws JMSException {
//						return session.createTextMessage(ms);
//					}
//				});
				log.error("==>MailException:", e);
			}
		} catch (Exception e) {
			log.error("==>", e);
		}
	}
	
}

最後是消費端的啟動測試類，MQConsumer，

public class MQConsumer {
	
	private static final Log log = LogFactory.getLog(MQConsumer.class);

	public static void main(String[] args) {
		try {
			ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring/spring-context.xml");
			context.start();
			System.out.println("consumer is starting.....");
		} catch (Exception e) {
			log.error("==>MQ context start error:", e);
			System.exit(0);
		}
	}
}

接下來就可以啟動兩個測試類，進行測試啦，首先啟動消費端的程式碼，在這裡插入圖片描述

此時，消費者成功啟動，訊息監聽器啟動，等待接收訊息，再啟動生產者，在這裡插入圖片描述

此時，生產者也啟動成功，並且成功傳送一條訊息到訊息佇列中，此時再看生產者端控制檯輸出的結果，在這裡插入圖片描述

再看管控臺，在這裡插入圖片描述

大家還可以繼續在此基礎上嘗試在ssm專案中整合activemq，原理基本類似，最後感謝大家觀看！！！

spring整合使用activemq

Spring整合activeMQ消息隊列

SPRING 整合 activemq 的 topic 模式

spring整合activemq傳遞物件注意事項

淺談jms之（通過spring整合activeMQ實現jms）例項

spring整合activemq衝突問題

Spring整合ActiveMQ教程

ActiveMQ學習（四）Spring整合ActiveMQ

Spring整合activeMq（一）:點對點模式

spring整合activemq傳送MQ訊息[queue模式]例項

spring整合activemq傳送MQ訊息[Topic模式]例項

spring 整合 activeMQ非同步傳送郵件

Spring整合ActiveMQ（topic 詳細配置步驟）

Spring整合ActiveMQ

Spring整合ActiveMQ-序列化的 Java物件

Spring整合activemq，實現單消費者消費

ActiveMQ學習總結（二）——spring整合ActiveMQ

深入淺出JMS(四)--Spring和ActiveMQ整合的完整實例

Spring mvc4 + ActiveMQ 整合

Spring和ActiveMQ整合的完整實例

消息中間件ActiveMQ及Spring整合JMS的介紹

spring整合使用activemq

相關推薦