spring整合JMS實(shí)現(xiàn)同步收發(fā)消息(基于ActiveMQ的實(shí)現(xiàn))
更新時(shí)間：2017年10月30日 11:07:17 作者：牛奮lch
本篇文章主要介紹了spring整合JMS實(shí)現(xiàn)同步收發(fā)消息(基于ActiveMQ的實(shí)現(xiàn))，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下
本文介紹了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文件如下：
<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為例來說明，配置如下：
<!-- 真正可以產(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é)約開銷。配置如下：
<!-- 真正可以產(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)的，配置如下：
<!-- 配置生產(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類來實(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ì)列中推送消息，接口定義如下：
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)聽的方式實(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來判斷下消息類型
*/
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è)用來表示目的地的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è)試代碼：
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!
再來看下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í)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
Java之SpringBoot集成ActiveMQ消息中間件案例講解
這篇文章主要介紹了Java之SpringBoot集成ActiveMQ消息中間件案例講解,本篇文章通過簡(jiǎn)要的案例,講解了該項(xiàng)技術(shù)的了解與使用,以下就是詳細(xì)內(nèi)容,需要的朋友可以參考下2021-07-07
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)查找替換功能，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-03-03
Java中Scanner類基礎(chǔ)使用、可能遇到的問題及注意事項(xiàng)
Scanner類是一個(gè)用于Scanner指的是java.util包下的Scanner類,可以接收控制臺(tái)輸入的數(shù)據(jù),這篇文章主要介紹了Java中Scanner類基礎(chǔ)使用、可能遇到的問題及注意事項(xiàng)的相關(guān)資料,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下2025-04-04
這篇文章主要介紹了詳解MyBatisPlus邏輯刪除與唯一索引沖突問題，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-03-03
SpringMVC?RESTFul實(shí)體類創(chuàng)建及環(huán)境搭建
這篇文章主要為大家介紹了SpringMVC?RESTFul實(shí)體類創(chuàng)建及環(huán)境搭建詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2022-05-05
這篇文章主要介紹了IDEA中解決 git pull 沖突的方法，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-07-07
SpringBoot Bean被加載時(shí)進(jìn)行控制
很多時(shí)候我們需要根據(jù)不同的條件在容器中加載不同的Bean，或者根據(jù)不同的條件來選擇是否在容器中加載某個(gè)Bean，這就是Bean的加載控制，一般我們可以通過編程式或注解式兩種不同的方式來完成Bean的加載控制2023-02-02
Spring Boot實(shí)現(xiàn)qq郵箱驗(yàn)證碼注冊(cè)和登錄驗(yàn)證功能
這篇文章主要給大家介紹了關(guān)于Spring Boot實(shí)現(xiàn)qq郵箱驗(yàn)證碼注冊(cè)和登錄驗(yàn)證功能的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-12-12
Java程序?qū)崿F(xiàn)導(dǎo)出Excel的方法(支持IE低版本)
下面小編就為大家?guī)硪黄狫ava程序?qū)崿F(xiàn)導(dǎo)出Excel的方法(支持IE低版本)。小編覺得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2016-07-07
這篇文章主要介紹了SpringAop實(shí)現(xiàn)操作日志記錄的方法，幫助大家更好的理解和使用SpringAop，感興趣的朋友可以了解下2020-12-12