spring整合JMS實(shí)現(xiàn)同步收發(fā)消息(基于ActiveMQ的實(shí)現(xiàn))
本文介紹了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的管理頁(yè)面，用戶名和密碼默認(rèn)都是admin，如下：
2. 新建一個(gè)Maven工程，并配置pom文件如下：
<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提供的連接工廠只是用來(lái)實(shí)現(xiàn)管理的，并不是真正連接MQ的，真正的連接工廠需要具體的MQ廠商提供，下面我們以ActiveMQ為例來(lái)說(shuō)明，配置如下：
<!-- 真正可以產(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，通過(guò)連接工廠池，可以將Connection，Session等都放在池里面，用的時(shí)候直接返回池里面的內(nèi)容，無(wú)需臨時(shí)建立連接，節(jié)約開(kāi)銷。配置如下：
<!-- 真正可以產(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>
<!-- 通過(guò)往PooledConnectionFactory注入一個(gè)ActiveMQConnectionFactory可以用來(lái)將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ā)送和接收消息，都是通過(guò)JmsTemplate來(lái)實(shí)現(xiàn)的，配置如下：
<!-- 配置生產(chǎn)者：配置好ConnectionFactory之后我們就需要配置生產(chǎn)者。生產(chǎn)者負(fù)責(zé)產(chǎn)生消息并發(fā)送到JMS服務(wù)器，這通常對(duì)應(yīng)的是我們的一個(gè)業(yè)務(wù)邏輯服務(wù)實(shí)現(xiàn)類。 但是我們的服務(wù)實(shí)現(xiàn)類是怎么進(jìn)行消息的發(fā)送的呢？這通常是利用Spring為我們提供的JmsTemplate類來(lái)實(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)者主要用來(lái)生產(chǎn)消息，并向目的隊(duì)列中推送消息，接口定義如下：
public interface ProducerService {
void sendMessage(Destination destination, final String message);
}
實(shí)現(xiàn)類代碼如下：
@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)聽(tīng)的方式實(shí)現(xiàn)，接口定義如下：
public interface ConsumerService {
String receiveMessage(Destination destination, Destination replyDestination);
}
實(shí)現(xiàn)類代碼如下：
@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來(lái)判斷下消息類型
*/
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)目完整的配置文件：
<?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è)用來(lái)表示目的地的Destination接口，它里面沒(méi)有任何方法定義，只是用來(lái)做一個(gè)標(biāo)識(shí)而已。當(dāng)我們?cè)谑褂肑msTemplate進(jìn)行消息發(fā)送時(shí)沒(méi)有指定destination的時(shí)候?qū)⑹褂媚J(rèn)的Destination。 默認(rèn)Destination可以通過(guò)在定義jmsTemplate bean對(duì)象時(shí)通過(guò)屬性defaultDestination或defaultDestinationName來(lái)進(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è)試代碼：
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è)試代碼：
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é)果如下：
生產(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!
再來(lái)看下ActiveMQ的管理頁(yè)面的結(jié)果：
從管理頁(yè)面中可以看到，生產(chǎn)者生產(chǎn)了消息，并且入隊(duì)列了，同時(shí)消費(fèi)者也消費(fèi)了消息，并將回復(fù)消息放到了回復(fù)隊(duì)列中，測(cè)試成功。
但是這種同步取消息的方式有個(gè)缺點(diǎn)，每次只會(huì)取一條消息消費(fèi)，取完之后就會(huì)一直阻塞，下面來(lái)測(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)聽(tīng)模式，測(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管理頁(yè)面如下：
消費(fèi)者消費(fèi)一條消息：
收到生產(chǎn)者的消息:hello acticeMQ:my name is chhliu!
hello acticeMQ:my name is chhliu!
消費(fèi)者消費(fèi)消息后，ActiveMQ管理頁(yè)面如下：
從上面的對(duì)比中，我們可以看出來(lái)，同步模式下，消費(fèi)者消費(fèi)消息時(shí)，是逐條消費(fèi)，每次只消費(fèi)一條消息。
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
這篇文章主要為大家詳細(xì)介紹了JavaGUI模仿QQ聊天功能完整版，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-07-07
Java實(shí)現(xiàn)兩個(gè)日期相減等于天數(shù)
這篇文章主要介紹了Java兩個(gè)日期相減等于天數(shù)的實(shí)現(xiàn)方式，本文通過(guò)兩種方式結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2023-09-09
SpringCloud?Gateway詳細(xì)分析實(shí)現(xiàn)負(fù)載均衡與熔斷和限流
這篇文章主要介紹了SpringCloud?Gateway實(shí)現(xiàn)路由轉(zhuǎn)發(fā)，負(fù)載均衡，熔斷和限流，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2022-07-07
本文詳細(xì)介紹了MAVEN_HOME的配置步驟、Path環(huán)境變量的設(shè)置、檢測(cè)配置是否成功的方法、修改默認(rèn)的maven依賴包下載路徑以及配置阿里鏡像倉(cāng)庫(kù)的路徑,同時(shí)分享了作者在配置過(guò)程中遇到的問(wèn)題,如命令不識(shí)別、版本不匹配等,并提供了解決方案2024-09-09
springboot集成redis實(shí)現(xiàn)簡(jiǎn)單秒殺系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了springboot集成redis實(shí)現(xiàn)簡(jiǎn)單秒殺系統(tǒng)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-12-12
這篇文章主要介紹了SpringBoot整合Shiro和Redis的示例代碼，本文通過(guò)示例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2022-04-04
MyBatis-Plus通用枚舉自動(dòng)關(guān)聯(lián)注入的實(shí)現(xiàn)
本文主要介紹了MyBatis-Plus通用枚舉自動(dòng)關(guān)聯(lián)注入的實(shí)現(xiàn)，解決了繁瑣的配置，讓 mybatis 優(yōu)雅的使用枚舉屬性，感興趣的可以一起來(lái)了解一下2021-06-06
java導(dǎo)出Excel通用方法的實(shí)例詳解
這篇文章主要介紹了java導(dǎo)出Excel通用方法的實(shí)例詳解的相關(guān)資料,需要的朋友可以參考下2017-07-07
本篇文章給大家?guī)?lái)的內(nèi)容是關(guān)于Java中Hashtable集合的常用方法詳解，有一定的參考價(jià)值，有需要的朋友可以參考一下，希望對(duì)你有所幫助。下面我們就來(lái)學(xué)習(xí)一下吧2021-11-11
Java中使用正則表達(dá)式獲取網(wǎng)頁(yè)中所有圖片的路徑
這篇文章主要介紹了Java中使用正則表達(dá)式獲取網(wǎng)頁(yè)中所有圖片的路徑,本文直接給出實(shí)例代碼,需要的朋友可以參考下2015-06-06