本文介紹了spring整合jms實現同步收發消息(基于activemq的實現)，分享給大家，具體如下：
1. 安裝activemq
注意：jdk版本需要1.7及以上才行
到apache官方網站下載最新的activemq的安裝包，并解壓到本地目錄下，下載鏈接如下：http://activemq.apache.org/download.html，解壓后的目錄結構如下：
bin目錄結構如下：
如果我們是32位的機器，就雙擊win32目錄下的activemq.bat,如果是64位機器，則雙擊win64目錄下的activemq.bat,運行結果如下：
啟動成功！成功之后在瀏覽器輸入地址 http://127.0.0.1:8161/，可以看到activemq的管理頁面，用戶名和密碼默認都是admin，如下：
2. 新建一個maven工程，并配置pom文件如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
<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>com.chhliu.myself</groupid> <artifactid>activemq_start</artifactid> <version>0.0.1-snapshot</version> <packaging>jar</packaging> <name>activemq_start</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceencoding>utf-8</project.build.sourceencoding> <spring-version>3.2.5.release</spring-version> </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>${spring-version}</version> </dependency> <dependency> <groupid>org.springframework</groupid> <artifactid>spring-jms</artifactid> <version>${spring-version}</version> </dependency> <dependency> <groupid>org.springframework</groupid> <artifactid>spring-test</artifactid> <version>${spring-version}</version> </dependency> <dependency> <groupid>javax.annotation</groupid> <artifactid>jsr250-api</artifactid> <version>1.0</version> </dependency> <dependency> <groupid>org.apache.activemq</groupid> <artifactid>activemq-all</artifactid> <version>5.13.3</version> </dependency> <dependency> <groupid>org.apache.commons</groupid> <artifactid>commons-pool2</artifactid> <version>2.0</version> </dependency> </dependencies> </project> |
3. 配置連接工廠(connectionfactory)
spring給我們提供了如下的連接工廠：
其中singleconnectionfactory保證每次返回的都是同一個連接，cachingconnectionfactory繼承了singleconnectionfactory，在保證同一連接的同時，增加了緩存的功能，可以緩存session以及生產者，消費者。當然，jms提供的連接工廠只是用來實現管理的，并不是真正連接mq的，真正的連接工廠需要具體的mq廠商提供，下面我們以activemq為例來說明，配置如下：
|
1
2
3
4
5
6
7
8
|
<!-- 真正可以產生connection的connectionfactory，由對應的 jms服務廠商提供 --> <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory"> <property name="brokerurl" value="tcp://localhost:61616" /> </bean> <bean id="connectionfactory" class="org.springframework.jms.connection.singleconnectionfactory"> <property name="targetconnectionfactory" ref="targetconnectionfactory" /> </bean> |
為了減少我們連接的資源消耗，activemq為我們提供了一個連接工廠管理池--pooledconnectionfactory，通過連接工廠池，可以將connection，session等都放在池里面，用的時候直接返回池里面的內容，無需臨時建立連接，節約開銷。配置如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
<!-- 真正可以產生connection的connectionfactory，由對應的 jms服務廠商提供 --> <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory"> <property name="brokerurl" value="tcp://localhost:61616" /> </bean> <!-- 通過往pooledconnectionfactory注入一個activemqconnectionfactory可以用來將connection，session和messageproducer池化這樣可以大大減少我們的資源消耗， --> <bean id="pooledconnectionfactory" class="org.apache.activemq.pool.pooledconnectionfactory"> <property name="connectionfactory" ref="targetconnectionfactory" /> <property name="maxconnections" value="10" /> </bean> <bean id="connectionfactory" class="org.springframework.jms.connection.singleconnectionfactory"> <property name="targetconnectionfactory" ref="pooledconnectionfactory" /> </bean> |
4. 配置jmstemplate
配置好連接工廠之后，就需要配置jms的jmstemplate，jmstemplate的作用和jdbctemplate類似，我們發送和接收消息，都是通過jmstemplate來實現的，配置如下：
|
1
2
3
4
5
6
|
<!-- 配置生產者：配置好connectionfactory之后我們就需要配置生產者。生產者負責產生消息并發送到jms服務器，這通常對應的是我們的一個業務邏輯服務實現類。 但是我們的服務實現類是怎么進行消息的發送的呢？這通常是利用spring為我們提供的jmstemplate類來實現的， 所以配置生產者其實最核心的就是配置進行消息發送的jmstemplate。對于消息發送者而言，它在發送消息的時候要知道自己該往哪里發， 為此，我們在定義jmstemplate的時候需要往里面注入一個spring提供的connectionfactory對象 --> <!-- spring提供的jms工具類，它可以進行消息發送、接收等 --> <bean id="jmstemplate" class="org.springframework.jms.core.jmstemplate"> <!-- 這個connectionfactory對應的是我們定義的spring提供的那個connectionfactory對象 --> <property name="connectionfactory" ref="connectionfactory" /> </bean> |
5. 生產者實現
配置完這些之后，我們就可以寫代碼實現生產者和消費者了，生產者主要用來生產消息，并向目的隊列中推送消息，接口定義如下：
|
1
2
3
|
public interface producerservice { void sendmessage(destination destination, final string message); } |
實現類代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
@service("producerserviceimpl") public class producerserviceimpl implements producerservice { /** * 注入jmstemplate */ @resource(name="jmstemplate") private jmstemplate jtemplate; /** * attention: * details：發送消息 * @author chhliu * 創建時間：2016-7-28 下午2:33:14 * @param destination * @param message */ @override public void sendmessage(destination receivedestination, final string message) { system.out.println("================生產者創建了一條消息=============="); jtemplate.send(receivedestination, new messagecreator() { @override public message createmessage(session session) throws jmsexception { return session.createtextmessage("hello acticemq:"+message); } }); } } |
6. 消費者實現
假設生產者已經創建了一條消息，并推送到了對應的隊列中，消費者需要從這個隊列中取出消息，并同時回復一條報文，自己已經收到了這條消息，為了測試回復報文的功能，我們下面會將回復報文放到另一個隊列中，此例使用同步接收消息的方式，而不是異步監聽的方式實現，接口定義如下：
|
1
2
3
|
public interface consumerservice { string receivemessage(destination destination, destination replydestination); } |
實現類代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
@service("consumerserviceimpl") public class consumerserviceimpl implements consumerservice { /** * 注入jmstemplate */ @resource(name="jmstemplate") private jmstemplate jtemplate; /** * attention: * details：接收消息，同時回復消息 * @author chhliu * 創建時間：2016-7-28 下午2:39:45 * @param destination * @return */ @override public string receivemessage(destination destination, destination replydestination) { /** * 接收消息隊列中的消息 */ message message = jtemplate.receive(destination); try { /** * 此處為了更好的容錯性，可以使用instanceof來判斷下消息類型 */ if(message instanceof textmessage){ string receivemessage = ((textmessage) message).gettext(); system.out.println("收到生產者的消息:"+receivemessage); /** * 收到消息之后，將回復報文放到回復隊列里面去 */ jtemplate.send(replydestination, new messagecreator() { @override public message createmessage(session session) throws jmsexception { return session.createtextmessage("消費者已經收到生產者的消息了，這是一條確認報文!"); } }); return receivemessage; } } catch (jmsexception e) { e.printstacktrace(); } return ""; } } |
生產者和消費者實現之后，我們要做的就是配置隊列了，下面給出項目完整的配置文件：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
|
<?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" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:cache="http://www.springframework.org/schema/cache" xmlns:jpa="http://www.springframework.org/schema/data/jpa" xsi:schemalocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-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/cache http://www.springframework.org/schema/cache/spring-cache-3.2.xsd http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa-1.3.xsd"> <!-- 掃描注解包 --> <context:annotation-config /> <context:component-scan base-package="com.chhliu.myself.activemq.start"></context:component-scan> <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory"> <property name="brokerurl" value="tcp://localhost:61616" /> </bean> <bean id="pooledconnectionfactory" class="org.apache.activemq.pool.pooledconnectionfactory"> <property name="connectionfactory" ref="targetconnectionfactory" /> <property name="maxconnections" value="10" /> </bean> <bean id="connectionfactory" class="org.springframework.jms.connection.singleconnectionfactory"> <property name="targetconnectionfactory" ref="pooledconnectionfactory" /> </bean> <bean id="jmstemplate" class="org.springframework.jms.core.jmstemplate"> <property name="connectionfactory" ref="connectionfactory" /> </bean> <!-- 在真正利用jmstemplate進行消息發送的時候，我們需要知道消息發送的目的地，即destination。 在jms中有一個用來表示目的地的destination接口，它里面沒有任何方法定義，只是用來做一個標識而已。當我們在使用jmstemplate進行消息發送時沒有指定destination的時候將使用默認的destination。 默認destination可以通過在定義jmstemplate bean對象時通過屬性defaultdestination或defaultdestinationname來進行注入， defaultdestinationname對應的就是一個普通字符串 --> <!--這個是隊列目的地，點對點的 --> <bean id="queuedestination" class="org.apache.activemq.command.activemqqueue"> <constructor-arg> <value>ntf_mock_input</value> </constructor-arg> </bean> <!--這個是回復隊列，點對點的 --> <bean id="responsequeue" class="org.apache.activemq.command.activemqqueue"> <constructor-arg> <value>ntf_mock_output</value> </constructor-arg> </bean> </beans> |
到這里，所有的代碼和配置文件就都整好了，下面就是進行測試，測試代碼如下：
生產者測試代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
package com.chhliu.myself.activemq.start.sync; import javax.annotation.resource; import javax.jms.destination; import org.junit.test; import org.junit.runner.runwith; import org.springframework.test.context.contextconfiguration; import org.springframework.test.context.junit4.springjunit4classrunner; @runwith(springjunit4classrunner.class) @contextconfiguration(locations = { "classpath:applicationcontext.xml" }) public class syncproduceractivemqtest { @resource(name="producerserviceimpl") private producerservice pservice; @resource(name="queuedestination") private destination receivequeue; @test public void producertest(){ pservice.sendmessage(receivequeue, "my name is chhliu!"); } } |
消費者測試代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
|
package com.chhliu.myself.activemq.start.sync; import javax.annotation.resource; import javax.jms.destination; import org.junit.test; import org.junit.runner.runwith; import org.springframework.test.context.contextconfiguration; import org.springframework.test.context.junit4.springjunit4classrunner; @runwith(springjunit4classrunner.class) @contextconfiguration(locations = { "classpath:applicationcontext.xml" }) public class syncconsumeractivemqtest { @resource(name="consumerserviceimpl") private consumerservice cservice; @resource(name="queuedestination") private destination receivequeue; @resource(name="responsequeue") private destination replyqueue; @test public void producertest(){ string result = cservice.receivemessage(receivequeue, replyqueue); system.out.println(result); } } |
測試結果如下：
|
1
2
3
4
5
|
生產者測試結果： ================生產者創建了一條消息============== 消費者測試結果： 收到生產者的消息:hello acticemq:my name is chhliu! hello acticemq:my name is chhliu! |
再來看下activemq的管理頁面的結果：
從管理頁面中可以看到，生產者生產了消息，并且入隊列了，同時消費者也消費了消息，并將回復消息放到了回復隊列中，測試成功。
但是這種同步取消息的方式有個缺點，每次只會取一條消息消費，取完之后就會一直阻塞，下面來測試一下：首先讓生產者再生產5條消息，然后運行消費者程序，發現會只消費一條消息，除非我們在消費者程序里面加while(true)，一直輪詢隊列，這種實現方式不僅耗內存，效率也不是很高，后面，我們會對這種方式進行改進，使用異步監聽模式，測試效果如下：
生產者創建了5條消息：
=======生產者創建了一條消息========
=======生產者創建了一條消息========
=======生產者創建了一條消息========
=======生產者創建了一條消息========
======生產者創建了一條消息=========
activemq管理頁面如下：
消費者消費一條消息：
收到生產者的消息:hello acticemq:my name is chhliu!
hello acticemq:my name is chhliu!
消費者消費消息后，activemq管理頁面如下：
從上面的對比中，我們可以看出來，同步模式下，消費者消費消息時，是逐條消費，每次只消費一條消息。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：http://blog.csdn.net/liuchuanhong1/article/details/52057711
