今天就跟大家聊聊有關(guān)Java中的Spring消息隊(duì)列如何實(shí)現(xiàn)整合ActiveMq，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。
創(chuàng)新互聯(lián)公司是網(wǎng)站建設(shè)專家,致力于互聯(lián)網(wǎng)品牌建設(shè)與網(wǎng)絡(luò)營(yíng)銷，專業(yè)領(lǐng)域包括做網(wǎng)站、成都網(wǎng)站制作、電商網(wǎng)站制作開(kāi)發(fā)、小程序制作、微信營(yíng)銷、系統(tǒng)平臺(tái)開(kāi)發(fā)，與其他網(wǎng)站設(shè)計(jì)及系統(tǒng)開(kāi)發(fā)公司不同，我們的整合解決方案結(jié)合了恒基網(wǎng)絡(luò)品牌建設(shè)經(jīng)驗(yàn)和互聯(lián)網(wǎng)整合營(yíng)銷的理念，并將策略和執(zhí)行緊密結(jié)合，且不斷評(píng)估并優(yōu)化我們的方案，為客戶提供全方位的互聯(lián)網(wǎng)品牌整合方案！
1、概述
首先和大家一起回顧一下Java 消息服務(wù)，在我之前的博客《Java消息隊(duì)列-JMS概述》中，我為大家分析了：
1.消息服務(wù)：一個(gè)中間件，用于解決兩個(gè)活多個(gè)程序之間的耦合，底層由Java 實(shí)現(xiàn)。
2.優(yōu)勢(shì)：異步、可靠
3.消息模型：點(diǎn)對(duì)點(diǎn)，發(fā)布/訂閱
4.JMS中的對(duì)象
然后在另一篇博客《Java消息隊(duì)列-ActiveMq實(shí)戰(zhàn)》中，和大家一起從0到1的開(kāi)啟了一個(gè)ActiveMq 的項(xiàng)目，在項(xiàng)目開(kāi)發(fā)的過(guò)程中，我們對(duì)ActiveMq有了一定的了解：
1.多種語(yǔ)言和協(xié)議編寫(xiě)客戶端。語(yǔ)言: Java, C, C++, C#, Ruby, Perl, Python, PHP。應(yīng)用協(xié)議: OpenWire,Stomp REST,WS Notification,XMPP,AMQP
2.完全支持JMS1.1和J2EE 1.4規(guī)范 (持久化,XA消息,事務(wù))
3.對(duì)Spring的支持,ActiveMQ可以很容易內(nèi)嵌到使用Spring的系統(tǒng)里面去,而且也支持Spring2.0的特性
4.通過(guò)了常見(jiàn)J2EE服務(wù)器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的測(cè)試,其中通過(guò)JCA 1.5 resource adaptors的配置,可以讓ActiveMQ可以自動(dòng)的部署到任何兼容J2EE 1.4 商業(yè)服務(wù)器上
5.支持多種傳送協(xié)議:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
6.支持通過(guò)JDBC和journal提供高速的消息持久化
7.從設(shè)計(jì)上保證了高性能的集群,客戶端-服務(wù)器,點(diǎn)對(duì)點(diǎn)
8.支持Ajax
9.支持與Axis的整合
10.可以很容易得調(diào)用內(nèi)嵌JMS provider,進(jìn)行測(cè)試
在接下來(lái)的這篇博客中，我會(huì)和大家一起來(lái)整合Spring 和ActiveMq，這篇博文,我們基于Spring+JMS+ActiveMQ+Tomcat，實(shí)現(xiàn)了Point-To-Point的異步隊(duì)列消息和PUB/SUB（發(fā)布/訂閱）模型，簡(jiǎn)單實(shí)例，不包含任何業(yè)務(wù)。
2、目錄結(jié)構(gòu)
2.1 項(xiàng)目目錄
IDE選擇了IDEA（建議大家使用），為了避免下載jar 的各種麻煩，底層使用maven搭建了一個(gè)項(xiàng)目，整合了Spring 和ActiveMq
2.2 pom.xml
4.0.0
Crawl-Page
Crawl-Page
war
1.0-SNAPSHOT
Crawl-Page Maven Webapp
http://maven.apache.org
4.1.8.RELEASE
junit
junit
4.10
test
jstl
jstl
1.2
javax.servlet
servlet-api
provided
2.5
org.springframework
spring-core
${springframework}
org.springframework
spring-context
${springframework}
org.springframework
spring-tx
${springframework}
org.springframework
spring-webmvc
${springframework}
org.springframework
spring-jms
${springframework}
org.apache.xbean
xbean-spring
3.16
org.apache.activemq
activemq-core
5.7.0
org.apache.activemq
activemq-pool
5.12.1
commons-httpclient
commons-httpclient
3.1
Crawl-Page
org.apache.tomcat.maven
tomcat7-maven-plugin
8080
/
因?yàn)檫@里pom.xml 文件有點(diǎn)長(zhǎng)，就不展開(kāi)了。
我們可以看到其實(shí)依賴也就幾個(gè)，1、Spring 核心依賴 2、ActiveMq core和pool（這里如果同學(xué)們選擇導(dǎo)入jar，可以直接導(dǎo)入我們上一篇博客中說(shuō)道的那個(gè)activemq-all 這個(gè)jar包）3、java servlet 相關(guān)依賴
這里面我們選擇的ActiveMq pool 的依賴版本會(huì)和之后的dtd 有關(guān)系，需要版本對(duì)應(yīng)，所以同學(xué)們等下配置activemq 文件的時(shí)候，需要注意dtd 版本選擇
2.3 web.xml
web.xml 也大同小異，指定Spring 配置文件，springMvc 命名，編碼格式
<?xml version="1.0" encoding="UTF-8"?>
Archetype Created Web Application
contextConfigLocation
classpath:applicationContext*.xml;
org.springframework.web.context.ContextLoaderListener
springMVC
org.springframework.web.servlet.DispatcherServlet
contextConfigLocation
classpath:spring-mvc.xml
1
springMVC
/
characterEncodingFilter
org.springframework.web.filter.CharacterEncodingFilter
encoding
UTF-8
forceEncoding
true
characterEncodingFilter
/*
2.4 SpringMvc 和applicationContext.xml
這里面的SpringMVC沒(méi)什么特別，有需要的同學(xué)可以參考一下：
<?xml version="1.0" encoding="UTF-8"?>
applicationContext.xml 主要使用來(lái)裝載Bean，我們項(xiàng)目中并沒(méi)有什么特別的Java Bean，因此只用來(lái)指出包掃描路徑：
<?xml version="1.0" encoding="UTF-8"?>
2.5 applicationContext-ActiveMQ.xml
<?xml version="1.0" encoding="UTF-8"?>
Jaycekon
這里和大家講解一下這個(gè)配置文件，如果大家能夠從上述配置文件中看懂，可以跳過(guò)。同學(xué)們也可以在ActiveMQ官網(wǎng)中的查看。
1、ActiveMq 中的DTD，我們?cè)诼暶飨嚓P(guān)配置之前，我們需要先導(dǎo)入ActiveMq 中的DTD，不然Spring 并不理解我們的標(biāo)簽是什么意思。
我們?cè)趐om.xml 文件中有配置了activemq 的版本依賴我們這里的版本，需要和依賴的版本一樣，不然是找不到相關(guān)的dtd
2、amq:connectionFactory：很直白的一個(gè)配置項(xiàng)，用于配置我們鏈接工廠的地址和用戶名密碼，這里需要注意的是選擇tcp連接而不是http連接
3、jmsTemplate：比較重要的一個(gè)配置，這里指定了連接工廠，默認(rèn)消息發(fā)送目的地，還有連接時(shí)長(zhǎng)，發(fā)布消息的方式
3、項(xiàng)目結(jié)構(gòu)
3.1 ProducerService
package com.Jayce.Service;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
/**
* Created by Administrator on 2017/1/5.
*/
@Service
public class ProducerService {
@Resource(name="jmsTemplate")
private JmsTemplate jmsTemplate;
public void sendMessage(Destination destination,final String msg){
System.out.println(Thread.currentThread().getName()+" 向隊(duì)列"+destination.toString()+"發(fā)送消息---------------------->"+msg);
jmsTemplate.send(destination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
public void sendMessage(final String msg){
String destination = jmsTemplate.getDefaultDestinationName();
System.out.println(Thread.currentThread().getName()+" 向隊(duì)列"+destination+"發(fā)送消息---------------------->"+msg);
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
});
}
}將消息生產(chǎn)者做成一個(gè)服務(wù)，當(dāng)我們需要發(fā)送消息的時(shí)候，只需要調(diào)用ProducerService實(shí)例中的sendMessage 方法就可以向默認(rèn)目的發(fā)送一個(gè)消息。
這里提供了兩個(gè)發(fā)送方式，一個(gè)是發(fā)送到默認(rèn)的目的地，一個(gè)是根據(jù)目的地發(fā)送消息。
3.2 ConsumerService
package com.Jayce.Service;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;
/**
* Created by Administrator on 2017/1/5.
*/
@Service
public class ConsumerService {
@Resource(name="jmsTemplate")
private JmsTemplate jmsTemplate;
public TextMessage receive(Destination destination){
TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
try{
System.out.println("從隊(duì)列" + destination.toString() + "收到了消息：\t"
+ textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
return textMessage;
}
}因?yàn)槲覀冺?xiàng)目中并沒(méi)有什么業(yè)務(wù)，所以的話對(duì)消息的處理也就是打印輸出。我們只需要調(diào)用jmsTemplate中的 receive 方法，就可以從里面獲取到一條消息。
再和我們上一篇博客對(duì)比一下，上一篇博客中，我們接受到信息之后需要手動(dòng)確認(rèn)事務(wù)，這樣ActiveMQ中才會(huì)確定這條消息已經(jīng)被正確讀取了。而整合了Spring之后，事務(wù)將由Spring 來(lái)管理。
3.3 MessageController
package com.Jayce.Controller;
import com.Jayce.Service.ConsumerService;
import com.Jayce.Service.ProducerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;
import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.TextMessage;
/**
* Created by Administrator on 2017/1/5.
*/
@Controller
public class MessageController {
private Logger logger = LoggerFactory.getLogger(MessageController.class);
@Resource(name = "demoQueueDestination")
private Destination destination;
//隊(duì)列消息生產(chǎn)者
@Resource(name = "producerService")
private ProducerService producer;
//隊(duì)列消息消費(fèi)者
@Resource(name = "consumerService")
private ConsumerService consumer;
@RequestMapping(value = "/SendMessage", method = RequestMethod.POST)
@ResponseBody
public void send(String msg) {
logger.info(Thread.currentThread().getName()+"------------send to jms Start");
producer.sendMessage(msg);
logger.info(Thread.currentThread().getName()+"------------send to jms End");
}
@RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET)
@ResponseBody
public Object receive(){
logger.info(Thread.currentThread().getName()+"------------receive from jms Start");
TextMessage tm = consumer.receive(destination);
logger.info(Thread.currentThread().getName()+"------------receive from jms End");
return tm;
}
}控制層里面需要注入我們的生產(chǎn)者和消費(fèi)者（實(shí)際開(kāi)發(fā)中，生產(chǎn)者和消費(fèi)者肯定不會(huì)在同一個(gè)項(xiàng)目中的，不然就消息服務(wù)這個(gè)東西就沒(méi)有意義了）。
現(xiàn)在服務(wù)層和控制層都好了，接下來(lái)我們就進(jìn)行一個(gè)簡(jiǎn)單的測(cè)試
4、項(xiàng)目測(cè)試
4.1 啟動(dòng)ActiveMq
先確定你的ActiveMQ服務(wù)已經(jīng)開(kāi)啟。
4.2 啟動(dòng)項(xiàng)目
項(xiàng)目使用了Tomcat 插件，避免了本地再下載Tomcat的麻煩，有需要的同學(xué)可以使用一下。
org.apache.tomcat.maven
tomcat7-maven-plugin
8080
/
4.3 發(fā)送消息
這里用了Chrome 的一個(gè)插件PostMan 有興趣的同學(xué)可以了解一下，在Chrome 拓展程序中可以找到，避免了后端的同學(xué)去弄頁(yè)面！
我們發(fā)送了一個(gè)post 請(qǐng)求之后，看一下服務(wù)器的效果：
我們可以看到，已經(jīng)向隊(duì)列發(fā)送了一條消息。我們看一下ActiveMq現(xiàn)在的狀態(tài)：
我們可以看到，一條消息已經(jīng)成功發(fā)送到了ActiveMq中。
4.4 接收消息
使用get請(qǐng)求訪問(wèn)服務(wù)器后臺(tái)：
服務(wù)的輸出：
ActiveMq服務(wù)器狀態(tài)：
我們可以看到，消費(fèi)者已經(jīng)消費(fèi)了一條信息，并且沒(méi)有斷開(kāi)與ActiveMq之間的鏈接。
4.5 監(jiān)聽(tīng)器
在實(shí)際項(xiàng)目中，我們很少會(huì)自己手動(dòng)去獲取消息，如果需要手動(dòng)去獲取消息，那就沒(méi)有必要使用到ActiveMq了，可以用一個(gè)redis 就足夠了。
不能手動(dòng)去獲取消息，那么我們就可以選擇使用一個(gè)監(jiān)聽(tīng)器來(lái)監(jiān)聽(tīng)是否有消息到達(dá)，這樣子可以很快的完成對(duì)消息的處理。
4.5.1 applicationContext-ActiveMQ.xml 配置
在上面的配置文件中，我們已經(jīng)默認(rèn)的添加了這段監(jiān)聽(tīng)器的配置文件，如果同學(xué)們不想使用這個(gè)監(jiān)聽(tīng)器，可以直接注釋掉。
4.5.2 MessageListener
我們需要?jiǎng)?chuàng)建一個(gè)類實(shí)現(xiàn)MessageListener 接口：
package com.Jayce.Filter;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
/**
* Created by Administrator on 2017/1/5.
*/
public class QueueMessageListener implements MessageListener {
public void onMessage(Message message) {
TextMessage tm = (TextMessage) message;
try {
System.out.println("QueueMessageListener監(jiān)聽(tīng)到了文本消息：\t"
+ tm.getText());
//do something ...
} catch (JMSException e) {
e.printStackTrace();
}
}
}實(shí)現(xiàn)接口的onMessage 方法，我們將需要的業(yè)務(wù)操作在里面解決，這樣子，就完成了我們生產(chǎn)者-中間件-消費(fèi)者，這樣一個(gè)解耦的操作了。
4.5.3 測(cè)試
和上面一樣，使用postMan 發(fā)送post請(qǐng)求，我們可以看到控制臺(tái)里面，消息馬上就能打印出來(lái)：
再看看ActiveMQ服務(wù)器的狀態(tài)：
我們可以看到，使用監(jiān)聽(tīng)器的效果，和手動(dòng)接收消息的效果是一樣的。
這樣子一整個(gè)項(xiàng)目下來(lái)，我們已經(jīng)成功的整合了Spring和ActiveMQ。
4.6 壓力測(cè)試
這里其實(shí)也算不上什么壓力測(cè)試，在配置pom.xml文件的時(shí)候，大家有看到一個(gè) commons-httpclient 的依賴，接下來(lái)我們使用httpClient 不停的想服務(wù)器發(fā)送消息，看一下服務(wù)器解決消息的速度如何：
package com.Jaycekon.test;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.PostMethod;
import org.junit.Test;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Created by Administrator on 2017/1/5.
*/
public class Client {
@Test
public void test() {
HttpClient httpClient = new HttpClient();
new Thread(new Sender(httpClient)).start();
}
}
class Sender implements Runnable {
public static AtomicInteger count = new AtomicInteger(0);
HttpClient httpClient;
public Sender(HttpClient client) {
httpClient = client;
}
public void run() {
try {
System.out.println(Thread.currentThread().getName()+"---Send message-"+count.getAndIncrement());
PostMethod post = new PostMethod("http://127.0.0.1:8080/SendMessage");
post.addParameter("msg", "Hello world!");
httpClient.executeMethod(post);
System.out.println(Thread.currentThread().getName()+"---Send message Success-"+count.getAndIncrement());
} catch (IOException e) {
e.printStackTrace();
}
}
}這里面用了HttpClient 來(lái)向服務(wù)器發(fā)送Post 請(qǐng)求，然后計(jì)數(shù)輸出，有興趣的同學(xué)可以自己測(cè)試一下，可以多開(kāi)幾個(gè)線程，這里只開(kāi)了一個(gè)線程。
看完上述內(nèi)容，你們對(duì)Java中的Spring消息隊(duì)列如何實(shí)現(xiàn)整合ActiveMq有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝大家的支持。
新聞標(biāo)題：Java中的Spring消息隊(duì)列如何實(shí)現(xiàn)整合ActiveMq
分享地址：
http://weahome.cn/article/gjgcpg.html
重慶分公司
重慶分公司
