Apache-ActiveMQ整合Spring
1.1關于activemq
在J2EE規范中,JMS有一席之地,但JMS只是一種規范一種概念,落地地實施JMS有很多產品,其中apache-activemq是一種比較常用的產品.ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現，盡管JMS規范出臺已經是很久的事情了，但是JMS在當今的J2EE應用中間仍然扮演著特殊的地位。
2 準備工作
2.1 下載并安裝activemq
下載apache-activemq:http://activemq.apache.org/download.html
這里下載linux版本:apache-activemq-5.7.0-bin.tar.gz
把apache-activemq-5.7.0-bin.tar.gz解壓到一個目錄中,會有一個apache-activemq-5.7.0的文件夾,進入該文件夾的bin目錄,將看到activemq的腳本文件,執行該文件: ./activemq
若無法執行,可能是權限不夠,需要修改權限: chmod 755 activemq
執行該文件后在終端如果看到下面信息,者啟動成功
此時activemq線程已經運行,這時候不要關閉終端,否則activemq也會終止,如果想在后臺運行activemq,執行一下命令:
nohup ./activemq start
3 實施項目
3.1 創建項目
新建maven項目,并聲明項目依賴:
<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.0</modelVersion> <groupId>ActivityMQ_Spring</groupId> <artifactId>ActivityMQ_Spring</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>war</packaging> <name>ActivityMQ_Spring</name> <description/> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.10</version> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-context</artifactId> <version>4.1.6.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-jms</artifactId> <version>4.1.6.RELEASE</version> </dependency> <dependency> <groupId>org.springframework</groupId> <artifactId>spring-test</artifactId> <version>4.1.6.RELEASE</version> </dependency> <dependency> <groupId>javax.annotation</groupId> <artifactId>jsr250-api</artifactId> <version>1.0</version> </dependency> <dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-core</artifactId> <version>5.7.0</version> </dependency> </dependencies> <build> <plugins> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> <source>1.7</source> <target>1.7</target> </configuration> </plugin> <plugin> <artifactId>maven-war-plugin</artifactId> <version>2.2</version> <configuration> <version>3.0</version> <failOnMissingWebXml>false</failOnMissingWebXml> </configuration> </plugin> </plugins> </build> </project>
配置Spring框架:
<?xml version="1.0" encoding="UTF-8"?> <web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://java.sun.com/xml/ns/javaee" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd" id="WebApp_ID" version="3.0"> <display-name>ActivityMQ_Spring</display-name> <!-- Spring監聽器 --> <listener> <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class> </listener> <!-- 框架配置文件位置 --> <context-param> <param-name>contextConfigLocation</param-name> <param-value>classpath*:applicationContext-*.xml</param-value> </context-param> </web-app>
3.2 創建消息發送者:
package com.lin.activity.intf;public interface ProductService { public void sendMessage(javax.jms.Destination destination, final String message);
}</pre>
package com.lin.activity.service;import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session;
import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator;
import com.lin.activity.intf.ProductService;
/* 消息發送組件是一個POJO,通過借助Spring的JmsTemplate封裝發送目的destination和消息發送消息 @author lin */ public class ProductServiceImpl implements ProductService { private JmsTemplate jmsTemplate;
public void setJmsTemplate(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; }
@Override public void sendMessage(javax.jms.Destination destination,final String message) { /* 發送消息,這里使用一個內部類實現,內部類引用的外層類的屬性值需要被定義為final */ jmsTemplate.send(destination, new MessageCreator() { @Override public Message createMessage(Session session) throws JMSException { return session.createTextMessage(message); } }); } }</pre>
3.3 創建消息接受者:
package com.lin.activity.service;import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageListener; import javax.jms.TextMessage;
/* 定義處理消息的MessageListener 要定義處理消息的MessageListener我們只需要實現JMS規范中的MessageListener接口就可以了。 MessageListener接口中只有一個方法onMessage方法，當接收到消息的時候會自動調用該方法。 @author lin */ public class CostumerListener implements MessageListener {
/* 該方法監聽發送方是否發送消息,若發送者處理消息 */ @Override public void onMessage(Message message) {
TextMessage textMsg = (TextMessage) message; System.out.println("接受者接受到一個文本消息......"); try { System.out.println("消息內容:" + textMsg.getText()); } catch (JMSException e) { e.printStackTrace(); } }
}</pre>
3.4 實現測試組件
package com.lin.activity.test;import javax.jms.Destination;
import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.lin.activity.intf.ProductService;
/* Spring組件測試類 @author lin / //配置Spring組件運行時 @RunWith(SpringJUnit4ClassRunner.class) //配置測試組件的配置文件路徑,該路徑位于class或maven項目的resource文件夾下 @ContextConfiguration("/applicationContext-jms.xml") public class ActiveMQTest { //屬性自動綁定 @Autowired @Qualifier("queueDestination") private Destination destination; @Autowired private ProductService productService;
public Destination getDestination() { return destination; }
public void setDestination(Destination destination) { this.destination = destination; }
public ProductService getProductService() { return productService; }
public void setProductService(ProductService productService) { this.productService = productService; } //測試方法 @org.junit.Test public void testSendMessage() { for(int i=0;i<10;i++) { productService.sendMessage(destination, "發送第 "+i+" 條消息......"); } }
}</pre>
3.5 配置applicationContext*.xml文件管理依賴:
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="<!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 --> <property name="connectionFactory" ref="connectionFactory" /> </bean> <!-- 消息發送組件 --> <bean id="productServiceImpl" class="com.lin.activity.service.ProductServiceImpl"> <property name="jmsTemplate" ref="jmsTemplate"></property> </bean>
<!-- 真正可以產生Connection的ConnectionFactory，由對應的 JMS服務廠商提供 --> <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> <property name="brokerURL" value="tcp://localhost:61616" /> </bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory --> <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory"> <!-- 目標ConnectionFactory對應真實的可以產生JMS Connection的ConnectionFactory --> <property name="targetConnectionFactory" ref="targetConnectionFactory" /> </bean>
<!--這個是隊列目的地 --> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg> <value>queue</value> </constructor-arg> </bean> <!-- 消息監聽器 --> <bean id="costumerListener" class="com.lin.activity.service.CostumerListener" /> <!-- 消息監聽容器 --> <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="connectionFactory" /> <property name="destination" ref="queueDestination" /> <property name="messageListener" ref="costumerListener" /> </bean> <!-- 單元測試類 --> <!-- - <bean id="activeMQTest" class="com.lin.activity.test.ActiveMQTest"> <property name="destination" ref="queueDestination"></property> <property name="productService" ref="productServiceImpl"></property> </bean> --> </beans></pre>
在上面配置中,在配置一個MessageListenerContainer的時候有三個屬性必須指定，一個是表示從哪里監聽的ConnectionFactory；一個是表示監聽什么的Destination；一個是接收到消息以后進行消息處理的MessageListener。Spring一共為我們提供了兩種類型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
SimpleMessageListenerContainer會在一開始的時候就創建一個會話session和消費者Consumer，并且會使用標準的JMS MessageConsumer.setMessageListener()方法注冊監聽器讓JMS提供者調用監聽器的回調函數。它不會動態的適應運行時需要和參與外部的事務管理。兼容性方面，它非常接近于獨立的JMS規范，但一般不兼容Java EE的JMS限制。
大多數情況下我們還是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer會動態的適應運行時需要，并且能夠參與外部的事務管理。它很好的平衡了對JMS提供者要求低、先進功能如事務參與和兼容Java EE環境。
4 測試
以單元測試運行ActiveMQTest的方法
關閉activemq: ./activemq stop
該實例的目錄結構:
來自：http://my.oschina.net/u/2328736/blog/533863