本文介紹了spring整合jms實(shí)現(xiàn)同步收發(fā)消息(基于activemq的實(shí)現(xiàn))，分享給大家，具體如下：
1. 安裝activemq
注意：jdk版本需要1.7及以上才行
到apache官方網(wǎng)站下載最新的activemq的安裝包，并解壓到本地目錄下，下載鏈接如下：http://activemq.apache.org/download.html，解壓后的目錄結(jié)構(gòu)如下：
bin目錄結(jié)構(gòu)如下：
如果我們是32位的機(jī)器，就雙擊win32目錄下的activemq.bat,如果是64位機(jī)器，則雙擊win64目錄下的activemq.bat,運(yùn)行結(jié)果如下：
啟動(dòng)成功！成功之后在瀏覽器輸入地址 http://127.0.0.1:8161/，可以看到activemq的管理頁面，用戶名和密碼默認(rèn)都是admin，如下：
2. 新建一個(gè)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保證每次返回的都是同一個(gè)連接，cachingconnectionfactory繼承了singleconnectionfactory，在保證同一連接的同時(shí)，增加了緩存的功能，可以緩存session以及生產(chǎn)者，消費(fèi)者。當(dāng)然，jms提供的連接工廠只是用來實(shí)現(xiàn)管理的，并不是真正連接mq的，真正的連接工廠需要具體的mq廠商提供，下面我們以activemq為例來說明，配置如下：
|
1
2
3
4
5
6
7
8
|
<!-- 真正可以產(chǎn)生connection的connectionfactory，由對(duì)應(yīng)的 jms服務(wù)廠商提供 --> <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為我們提供了一個(gè)連接工廠管理池--pooledconnectionfactory，通過連接工廠池，可以將connection，session等都放在池里面，用的時(shí)候直接返回池里面的內(nèi)容，無需臨時(shí)建立連接，節(jié)約開銷。配置如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
<!-- 真正可以產(chǎn)生connection的connectionfactory，由對(duì)應(yīng)的 jms服務(wù)廠商提供 --> <bean id="targetconnectionfactory" class="org.apache.activemq.activemqconnectionfactory"> <property name="brokerurl" value="tcp://localhost:61616" /> </bean> <!-- 通過往pooledconnectionfactory注入一個(gè)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類似，我們發(fā)送和接收消息，都是通過jmstemplate來實(shí)現(xiàn)的，配置如下：
|
1
2
3
4
5
6
|
<!-- 配置生產(chǎn)者：配置好connectionfactory之后我們就需要配置生產(chǎn)者。生產(chǎn)者負(fù)責(zé)產(chǎn)生消息并發(fā)送到j(luò)ms服務(wù)器，這通常對(duì)應(yīng)的是我們的一個(gè)業(yè)務(wù)邏輯服務(wù)實(shí)現(xiàn)類。 但是我們的服務(wù)實(shí)現(xiàn)類是怎么進(jìn)行消息的發(fā)送的呢？這通常是利用spring為我們提供的jmstemplate類來實(shí)現(xiàn)的， 所以配置生產(chǎn)者其實(shí)最核心的就是配置進(jìn)行消息發(fā)送的jmstemplate。對(duì)于消息發(fā)送者而言，它在發(fā)送消息的時(shí)候要知道自己該往哪里發(fā)， 為此，我們?cè)诙xjmstemplate的時(shí)候需要往里面注入一個(gè)spring提供的connectionfactory對(duì)象 --> <!-- spring提供的jms工具類，它可以進(jìn)行消息發(fā)送、接收等 --> <bean id="jmstemplate" class="org.springframework.jms.core.jmstemplate"> <!-- 這個(gè)connectionfactory對(duì)應(yīng)的是我們定義的spring提供的那個(gè)connectionfactory對(duì)象 --> <property name="connectionfactory" ref="connectionfactory" /> </bean> |
5. 生產(chǎn)者實(shí)現(xiàn)
配置完這些之后，我們就可以寫代碼實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者了，生產(chǎn)者主要用來生產(chǎn)消息，并向目的隊(duì)列中推送消息，接口定義如下：
|
1
2
3
|
public interface producerservice { void sendmessage(destination destination, final string message); } |
實(shí)現(xiàn)類代碼如下：
|
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：發(fā)送消息 * @author chhliu * 創(chuàng)建時(shí)間：2016-7-28 下午2:33:14 * @param destination * @param message */ @override public void sendmessage(destination receivedestination, final string message) { system.out.println("================生產(chǎn)者創(chuàng)建了一條消息=============="); jtemplate.send(receivedestination, new messagecreator() { @override public message createmessage(session session) throws jmsexception { return session.createtextmessage("hello acticemq:"+message); } }); } } |
6. 消費(fèi)者實(shí)現(xiàn)
假設(shè)生產(chǎn)者已經(jīng)創(chuàng)建了一條消息，并推送到了對(duì)應(yīng)的隊(duì)列中，消費(fèi)者需要從這個(gè)隊(duì)列中取出消息，并同時(shí)回復(fù)一條報(bào)文，自己已經(jīng)收到了這條消息，為了測(cè)試回復(fù)報(bào)文的功能，我們下面會(huì)將回復(fù)報(bào)文放到另一個(gè)隊(duì)列中，此例使用同步接收消息的方式，而不是異步監(jiān)聽的方式實(shí)現(xiàn)，接口定義如下：
|
1
2
3
|
public interface consumerservice { string receivemessage(destination destination, destination replydestination); } |
實(shí)現(xiàn)類代碼如下：
|
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：接收消息，同時(shí)回復(fù)消息 * @author chhliu * 創(chuàng)建時(shí)間：2016-7-28 下午2:39:45 * @param destination * @return */ @override public string receivemessage(destination destination, destination replydestination) { /** * 接收消息隊(duì)列中的消息 */ message message = jtemplate.receive(destination); try { /** * 此處為了更好的容錯(cuò)性，可以使用instanceof來判斷下消息類型 */ if(message instanceof textmessage){ string receivemessage = ((textmessage) message).gettext(); system.out.println("收到生產(chǎn)者的消息:"+receivemessage); /** * 收到消息之后，將回復(fù)報(bào)文放到回復(fù)隊(duì)列里面去 */ jtemplate.send(replydestination, new messagecreator() { @override public message createmessage(session session) throws jmsexception { return session.createtextmessage("消費(fèi)者已經(jīng)收到生產(chǎn)者的消息了，這是一條確認(rèn)報(bào)文!"); } }); return receivemessage; } } catch (jmsexception e) { e.printstacktrace(); } return ""; } } |
生產(chǎn)者和消費(fèi)者實(shí)現(xiàn)之后，我們要做的就是配置隊(duì)列了，下面給出項(xiàng)目完整的配置文件：
|
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進(jìn)行消息發(fā)送的時(shí)候，我們需要知道消息發(fā)送的目的地，即destination。 在jms中有一個(gè)用來表示目的地的destination接口，它里面沒有任何方法定義，只是用來做一個(gè)標(biāo)識(shí)而已。當(dāng)我們?cè)谑褂胘mstemplate進(jìn)行消息發(fā)送時(shí)沒有指定destination的時(shí)候?qū)⑹褂媚J(rèn)的destination。 默認(rèn)destination可以通過在定義jmstemplate bean對(duì)象時(shí)通過屬性defaultdestination或defaultdestinationname來進(jìn)行注入， defaultdestinationname對(duì)應(yīng)的就是一個(gè)普通字符串 --> <!--這個(gè)是隊(duì)列目的地，點(diǎn)對(duì)點(diǎn)的 --> <bean id="queuedestination" class="org.apache.activemq.command.activemqqueue"> <constructor-arg> <value>ntf_mock_input</value> </constructor-arg> </bean> <!--這個(gè)是回復(fù)隊(duì)列，點(diǎn)對(duì)點(diǎn)的 --> <bean id="responsequeue" class="org.apache.activemq.command.activemqqueue"> <constructor-arg> <value>ntf_mock_output</value> </constructor-arg> </bean> </beans> |
到這里，所有的代碼和配置文件就都整好了，下面就是進(jìn)行測(cè)試，測(cè)試代碼如下：
生產(chǎn)者測(cè)試代碼：
|
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!"); } } |
消費(fèi)者測(cè)試代碼：
|
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); } } |
測(cè)試結(jié)果如下：
|
1
2
3
4
5
|
生產(chǎn)者測(cè)試結(jié)果： ================生產(chǎn)者創(chuàng)建了一條消息============== 消費(fèi)者測(cè)試結(jié)果： 收到生產(chǎn)者的消息:hello acticemq:my name is chhliu! hello acticemq:my name is chhliu! |
再來看下activemq的管理頁面的結(jié)果：
從管理頁面中可以看到，生產(chǎn)者生產(chǎn)了消息，并且入隊(duì)列了，同時(shí)消費(fèi)者也消費(fèi)了消息，并將回復(fù)消息放到了回復(fù)隊(duì)列中，測(cè)試成功。
但是這種同步取消息的方式有個(gè)缺點(diǎn)，每次只會(huì)取一條消息消費(fèi)，取完之后就會(huì)一直阻塞，下面來測(cè)試一下：首先讓生產(chǎn)者再生產(chǎn)5條消息，然后運(yùn)行消費(fèi)者程序，發(fā)現(xiàn)會(huì)只消費(fèi)一條消息，除非我們?cè)谙M(fèi)者程序里面加while(true)，一直輪詢隊(duì)列，這種實(shí)現(xiàn)方式不僅耗內(nèi)存，效率也不是很高，后面，我們會(huì)對(duì)這種方式進(jìn)行改進(jìn)，使用異步監(jiān)聽模式，測(cè)試效果如下：
生產(chǎn)者創(chuàng)建了5條消息：
=======生產(chǎn)者創(chuàng)建了一條消息========
=======生產(chǎn)者創(chuàng)建了一條消息========
=======生產(chǎn)者創(chuàng)建了一條消息========
=======生產(chǎn)者創(chuàng)建了一條消息========
======生產(chǎn)者創(chuàng)建了一條消息=========
activemq管理頁面如下：
消費(fèi)者消費(fèi)一條消息：
收到生產(chǎn)者的消息:hello acticemq:my name is chhliu!
hello acticemq:my name is chhliu!
消費(fèi)者消費(fèi)消息后，activemq管理頁面如下：
從上面的對(duì)比中，我們可以看出來，同步模式下，消費(fèi)者消費(fèi)消息時(shí)，是逐條消費(fèi)，每次只消費(fèi)一條消息。
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持服務(wù)器之家。
原文鏈接：http://blog.csdn.net/liuchuanhong1/article/details/52057711
