1、系統(tǒng)開發(fā)第 13 章消息中間件解決方案 JMS播客.程序員課程目標目標 1：理解消息中間件、JMS 等概念目標 2：掌握 JMS 點對點與發(fā)布訂閱模式的收發(fā)消息目標 3：掌握 SpringJms目標 4：完成商品審核導入索引庫目標 5：完成商品刪除移除索引庫目標 6: 完成商品審核生成商品詳細頁目標 7: 完成商品刪除完成刪除商品詳細頁1. JMS 入門1.1 消息中間件1.1.1系統(tǒng)模塊調用關系分析現(xiàn)在講已經(jīng)完成的功能模塊梳理如下：已經(jīng)完成了 5 個 web 模塊和 4 個服務模塊。其中運營商的調用關系最多，用到了商家商品服務、內容服務、搜索服務和頁面生成服務。這種模塊之間的依賴也稱之為耦合

2、。而耦合越多，之后的工作就越。那么如果改善系統(tǒng)模塊調用關系、減少模塊之間的耦合呢？接下來就介紹一種解決方案消息中間件。1.1.2消息中間件消息中間件利用高效可靠的消息傳遞機制進行無關的數(shù)據(jù)交流，并基于數(shù)據(jù)通信來進行分布式系統(tǒng)的集成。通過提供消息傳遞和消息排隊模型，它可以在分布式環(huán)境下擴展進的通信。對于消息中間件，常見的角色大致也就有 Producer（生產(chǎn)者）、Consumer（消費者）常見的消息中間件產(chǎn)品:（1）ActiveMQActiveMQ 是 Apache，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持 JMS1.1 和 J2EE 1.4 規(guī)范的 JMS Provi

3、der 實現(xiàn)用。在本次課程中介紹 ActiveMQ 的使（2）RabbitMQAMQP 協(xié)議的OpenStack 開源云實現(xiàn)，支持多種場景。淘寶的集群有使用它進行通訊，的通信組件，最先在金融行業(yè)得到運用。（3）ZeroMQ史上最快的消息隊列系統(tǒng)（4）KafkaApache 下的一個子項目 。特點：高吞吐，在一臺普通的服務器上既可以達到 10W/s的吞吐速率；完全的分布式系統(tǒng)。適合處理海量數(shù)據(jù)。1.1.3 改造系統(tǒng)模塊調用關系通過引入消息中間件 activeMQ,使得運營商系統(tǒng)與搜索服務、頁面生成服務解除了耦合。1.2 JMS 簡介1.2.1JMSJMS（Java Messaging Servi

4、ce）是 Java上有關面向消息中間件的技術規(guī)范，它便于消息系統(tǒng)中的 Java 應用程序進行消息交換,并且通過提供標準的產(chǎn)生、發(fā)送、接收消息的接口簡化企業(yè)應用的開發(fā)。JMS 本身只定義了一系列的接口規(guī)范，是一種與廠商無關的 API，用來統(tǒng)。它類似于 JDBC(java Database Connectivity)：這里，JDBC 是可以用來消息收發(fā)系許多不同關系數(shù)據(jù)庫的 API，而 JMS 則提供同樣與廠商無關的方法，以消息收發(fā)服務。許多廠商目前都支持 JMS，包括 IBM 的 MQSeries、BEA 的 Weblogic JMS service 和 Progress 的 SonicMQ，這

5、只是幾個例子。 JMS 使您能夠通過消息收發(fā)服務（有時稱為消息中介程序或路由器）從一個 JMS 客戶機向另一個 JML 客戶機發(fā)送消息。消息是 JMS 中的一種類型對象，由兩部分組成：報頭和消息主體。報頭由路由信息以及有關該消息的元數(shù)據(jù)組成。消息主體則攜帶著應用程序的數(shù)據(jù)或有效負載。JMS 定義了五種不同的消息正文格式，以及調用的消息類型，允許你發(fā)送并接收以一些不同形式的數(shù)據(jù)，提供現(xiàn)有消息格式的一些級別的兼容性。TextMessage-一個字符串對象MapMessage-一套名稱-值對ObjectMessage-一個序列化的 Java 對象BytesMessage-一個字節(jié)的數(shù)據(jù)流Stream

6、Message - Java 原始值的數(shù)據(jù)流1.2.2 JMS 消息傳遞類型對于消息的傳遞有兩種類型：一種是點對點的，即一個生產(chǎn)者和一個消費者一一對應；另一種是發(fā)布/ 訂閱模式，即一個生產(chǎn)者產(chǎn)生消息并進行發(fā)送后，可以由多個消費者進行接收。1.3ActiveMQ與安裝1.3.1：1.3.2 安裝（Linux）（1）將 apache-activemq-5.12.0-bar.gz 上傳至服務器（2）解壓此文件（3）為 apache-activemq-5.12.0 目錄賦權（4）進入 apache-activemq-5.12.0bin 目錄（5）賦與執(zhí)行權限od 755 activemq知識點小貼士l

7、inux 命令od 755 的意思文件所有者可讀可寫可執(zhí)行-7與文件所有者同屬一個用戶組的其他用戶可讀可執(zhí)行 -5od 是 Linux 下設置文件權限令，后面的數(shù)字表示不同用戶或用戶組的權限。一般是三個數(shù)字：第一個數(shù)字表示文件所有者的權限第二個數(shù)字表示與文件所有者同屬一個用戶組的其他用戶的權限第三個數(shù)字表示其它用戶組的權限。權限分為三種：讀（r=4），寫（w=2），執(zhí)行（x=1） 。 綜合起來還有可讀可執(zhí)行（rx=5=4+1）、可讀可寫（rw=6=4+2）、可讀可寫可執(zhí)行(rwx=7=4+2+1)。所以，od 755 設置用戶的權限為：od 777 apache-activemq-5.12.0

8、tarzxvfapache-activemq-5.12.0-bar.gz1.3.3 啟動出現(xiàn)下列提示表示成功！假設服務器地址為 192.168.25.135，打開瀏覽器輸入地址ActiveMQ 管理頁面點擊進入管理頁面輸入用戶名和均為 admin./activemq start3.其它用戶組可讀可執(zhí)行-5進入主界面點對點消息列表：列表各列信息含義如下：Number Of Pending Messages ：等待消費的消息 這個是當前未出隊列的數(shù)量。Number Of Consumers ：消費者 這個是消費者端的消費者數(shù)量Messages Enqueued ：進入隊列的消息 進入隊列的總數(shù)量,

9、包括出隊列的。Messages Dequeued ：出了隊列的消息 可以理解為是消費這消費掉的數(shù)量。2. JMS 入門小 Demo2.1 點對點模式點對點的模式主要建立在一個隊列上面，當連接一個列隊的時候，發(fā)送端不需要知道接收端是否正在接收，可以直接向 ActiveMQ 發(fā)送消息，發(fā)送的消息，將會先進入隊列中，如果有接收端在，則會發(fā)向接收端，如果沒有接收端接收，則會保存在 activemq 服務器，直到接收端接收消息，點對點的消息模式可以有多個發(fā)送端，多個接收端，但是一條消息，只會被一個接收端給接收到，哪個接收端先連上 ActiveMQ，則會先接收到，而后來的接收端則接收不到那條消息。2.1.

10、1 消息生產(chǎn)者（1）創(chuàng)建工程 jmsDemo ，引入依賴（2）創(chuàng)建類 QueueProducermain 方法代碼如下：/1.創(chuàng)建連接工廠ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(tcp:/192.168.25.135:61616);.apache.activemqactivemq-cnt5.13.4上述代碼中第 4 步創(chuàng)建 ses的兩個參數(shù)：第 1 個參數(shù)是否使用事務第 2 個參數(shù)消息的確認模式AUTO_ACKNOWLEDGE = 1自動確認NT_ACKNOWLEDGE = 2客戶端手動確認C/2.獲取

11、連接Connection connection = connectionFactory.createConnection();/3.啟動連接 connection.start();/4.獲取 ses(參數(shù) 1：是否啟動事務,參數(shù) 2：消息確認模式)Sesses= connection.createSes(false, Ses.AUTO_ACKNOWLEDGE);/5.創(chuàng)建隊列對象Queue queue = ses.createQueue(test-queue);/6.創(chuàng)建消息生產(chǎn)者MessageProducroducer = ses.createProducer(queue);/7.創(chuàng)建消息T

12、extMessage textMessage = ses.createTextMessage(歡迎來到神奇的世界);/8.發(fā)送消息 producer.send(textMessage);/9.關閉資源producer.close(); ses.close();connection.close();DUPS_OK_ACKNOWLEDGE = 3自動批量確認ED = 0事務提交并確認SES_TRAN運行后通過ActiveMQ 管理界面查詢2.1.2 消息消費者創(chuàng)建類 QueueConsumer ，main 方法代碼如下：/1.創(chuàng)建連接工廠ConnectionFactory connectionFa

13、ctory=new ActiveMQConnectionFactory(tcp:/192.168.25.135:61616);/2.獲取連接Connection connection = connectionFactory.createConnection();/3.啟動連接 connection.start();/4.獲取 ses(參數(shù) 1：是否啟動事務,參數(shù) 2：消息確認模式)Sesses= connection.createSes(false, Ses.AUTO_ACKNOWLEDGE);/5.創(chuàng)建隊列對象Queue queue = ses.createQueue(test-queue)

14、;/6.創(chuàng)建消息消費MessageConsumer consumer = ses.createConsumer(queue);執(zhí)行后看到控制臺輸出/7.消息consumer.setMessageListener(new MessageListener() public void onMessage(Message message) TextMessage textMessage=(TextMessage)message; try System.out.prln(接收到消息：+textMessage.getText(); catch (JMSException e) / TODO Auto-ge

15、nerated catch block e.prStackTrace(););/8.等待鍵盤輸入 System.in.read();/9.關閉資源consumer.close(); ses.close();connection.close();2.1.3 運試同時開啟 2 個以上的消費者，再次運行生產(chǎn)者，觀察每個消費者控制臺的輸出，會發(fā)現(xiàn)只有一個消費者會接收到消息。2.2 發(fā)布/訂閱模式2.2.1 消息生產(chǎn)者創(chuàng)建類 TopicProducer ，main 方法代碼如下：/1.創(chuàng)建連接工廠ConnectionFactory connectionFactory=new ActiveMQConnec

16、tionFactory(tcp:/192.168.25.135:61616);/2.獲取連接Connection connection = connectionFactory.createConnection();/3.啟動連接 connection.start();/4.獲取 ses(參數(shù) 1：是否啟動事務,參數(shù) 2：消息確認模式)Sesses= connection.createSes(false, Ses.AUTO_ACKNOWLEDGE);/5.創(chuàng)建對象Topic topic = ses.createTopic(test-topic);/6.創(chuàng)建消息生產(chǎn)者MessageProducro

17、ducer = ses.createProducer(topic);/7.創(chuàng)建消息TextMessage textMessage = ses.createTextMessage(歡迎來到神奇的世界);/8.發(fā)送消息 producer.send(textMessage);運行效果如下：2.2.2 消息消費者創(chuàng)建類 TopicConsumer ,main 方法代碼如下：/1.創(chuàng)建連接工廠ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(tcp:/192.168.25.135:61616);/2.獲取連接Connect

18、ion connection = connectionFactory.createConnection();/3.啟動連接 connection.start();/4.獲取 ses(參數(shù) 1：是否啟動事務,參數(shù) 2：消息確認模式)Sesses= connection.createSes(false, Ses.AUTO_ACKNOWLEDGE);/5.創(chuàng)建對象/Queue queue = ses.createQueue(test-queue); Topic topic = ses.createTopic(test-topic);/6.創(chuàng)建消息消費MessageConsumer consumer

19、= ses.createConsumer(topic);/9.關閉資源producer.close();ses.close(); connection.close();2.2.3 運試同時開啟 2 個以上的消費者，再次運行生產(chǎn)者，觀察每個消費者控制臺的輸出，會發(fā)現(xiàn)每個消費者會接收到消息。/7.消息consumer.setMessageListener(new MessageListener() public void onMessage(Message message) TextMessage textMessage=(TextMessage)message; try System.out.p

20、rln(接收到消息：+textMessage.getText(); catch (JMSException e) / TODO Auto-generated catch block e.prStackTrace(););/8.等待鍵盤輸入 System.in.read();/9.關閉資源consumer.close(); ses.close(); connection.close();3.Spring 整合 JMS3.1 點對點模式3.1.1 消息生產(chǎn)者（1）創(chuàng)建工程 springjms_producer，在 POM 文件中引入 SpringJms 、activeMQ 以及單元測試相關依賴（2

21、）在 src/main/resour下創(chuàng)建 spring 配置文件 applicationContext-jms-producer.xml（3）在 cn.demo 包下創(chuàng)建消息生產(chǎn)者類Componentpublic class QueueProducer Autowiredprivate JmsTemplate jmsTemplate;Autowiredprivate Destination queueTextDestination;/*發(fā)送文本消息param text*/public void sendTextMessage(final String text) jmsTemplate.se

22、nd(queueTextDestination, new MessageCreator() public Message createMessage(Sesses) throws JMSException return ses.createTextMessage(text);（4）單元測試在 src/test/java 創(chuàng)建測試類3.1.2 消息消費者（1）創(chuàng)建工程 springjms_consumer，在 POM 文件中引入依賴（同上一個工程）（2）創(chuàng)建配置文件 applicationContext-jms-consumer-queue.xml（4）創(chuàng)建測試類RunWith(SpringJU

23、nit4ClassRunner.class)ContextConfiguration(locations=classpapplicationContext-jms-consumer-queue.xm l)public class TestQueue public void testQueue()try System.in.read(); catch (IOException e) e.prStackTrace();Testpublic void onMessage(Message message) TextMessage=(TextMessage)message; try System.out

24、.prln(接收到消息：+.getText(); catch (JMSException e) e.prStackTrace();textMessagetextMessage3.2 發(fā)布/訂閱模式3.2.1 消息生產(chǎn)者（1）在工程 springjms_producer 的 applicationContext-jms-producer.xml 增加配置（2）創(chuàng)建生產(chǎn)者類Componentpublic class TopicProducer Autowiredprivate JmsTemplate jmsTemplate;Autowiredprivate Destination topicTex

25、tDestination;/*發(fā)送文本消息param text*/public void sendTextMessage(final String text) jmsTemplate.send(topicTextDestination, new MessageCreator() （3）編寫測試類import.junit.Test;import.junit.runner.RunWith;import.springframework.beans.factory.annoion.Autowired;import.springframework.test.context.ContextConfigur

26、ation;import.springframework.test.context.junit4.SpringJUnit4ClassRunner; import cn.demo.TopicProducer; RunWith(SpringJUnit4ClassRunner.class)ContextConfiguration(locations=classpapplicationContext-activemq-producer.xml )public class TestTopic Autowiredprivate TopicProducer topicProducer; Testpublic

27、 void sendTextQueue() topicProducer.sendTextMessage();public Message createMessage(Sesses) throws JMSException return ses.createTextMessage(text););3.2.2 消息消費者（1）在activemq-spring-consumer工程中創(chuàng)建配置文件applicationContext-jms-consumer-topic.xml（2）編寫測試類測試：同時運行三個消費者工程，在運行生產(chǎn)者工程，查看三個消費者工程的控制臺輸出。4.商品審核-導入 Solr

28、索引庫4.1 需求分析運用消息中間件 activeMQ 實現(xiàn)運營商與搜索服務的零耦合。運營商執(zhí)行商品審核后，向 activeMQ 發(fā)送消息（SKU 列表），搜索服務從 activeMQ 接收到消息并導入到 solr 索引庫。RunWith(SpringJUnit4ClassRunner.class)ContextConfiguration(locations=classpapplicationContext-jms-consumer-topic.xm l)public class TestTopic Testpublic void()try System.in.read(); catch (I

29、OException e) e.prStackTrace();testTopic4.2 消息生產(chǎn)者（運營商）4.2.1 解除耦合修改 pinyougou-manager-web，移除搜索服務接口依賴：GoodsController.java 中刪除調用搜索服務接口的相關代碼4.2.2 準備工作（1）修改 pinyougou-manager-web 的 pom.xml,引入依賴.apache.activemqactivemq-cnt5.13.4/Reference/private ItemSearchService itemSearchService;/itemSearchService.del

30、eteByGoodsIds(Arrays.asList(ids);/itemSearchService.importList(itemList);com.pinyougoupinyougou-search-erface0.0.3-SNAPSHOT（2）pinyougou-sellergoods-service 工程添加 applicationContext-jms-producer.xml改名為 spring-activemq.xml內容如下：（3）修改 web.xml4.2.3 代碼實現(xiàn)修改 pinyougou-manager-web 工程的GoodsController.javaAutow

31、iredprivate Destination queueSolrDestination;/用于發(fā)送 solr 導入的消息Autowiredprivate JmsTemplate jmsTemplate;RequestMap(/updateSus)public Result updateSus(Long ids,String sus)try goodsService.updateSus(ids, sus);/按照 SPU ID 查詢 SKU 列表(狀態(tài)為 1)contextConfigLocationclasspath:spring/spring-*.xmlif(sus.equals(1)/審

32、核通過List itemList = goodsService.findItemListByGoodsIdandSus(ids, sus);/調用搜索接口實現(xiàn)數(shù)據(jù)批量導入if(itemList.size()0)final String jsonString = JSON.toJSONString(itemList); jmsTemplate.send(queueSolrDestination, new MessageCreator() Overridepublic Message createMessage(Sesses) throws JMSExceptionreturn ses.creat

33、eTextMessage(jsonString););elseSystem.out.prln(沒有明細數(shù)據(jù));return new Result(true, 修改狀態(tài)成功); catch (Exception e) e.prStackTrace();return new Result(false, 修改狀態(tài)失敗);4.3 消息消費者（搜索服務）4.3.1 準備工作（1）修改 pinyougou-search-service ，在 pom.xml 中添加 activemq 依賴（2）添加 spring 配置文件 applicationContext-jms-consumer.xml.apache

34、.activemqactivemq-cnt5.13.44.3.2 代碼實現(xiàn)在 pinyougou-search-service 的 com.pinyougou.search.service.impl 新增類Componentpublic class ItemSearchListener implements MessageListener Autowiredprivate ItemSearchService itemSearchService;Overridepublic void onMessage(Message message) System.out.prln(接收到消息.);try T

35、extMessage textMessage=(TextMessage)message; String text = textMessage.getText();List list = JSON.parseArray(text,TbItem.class);for(TbItem item:list)System.out.prln(item.getId()+ +item.getTitle();Map specMap= JSON.parseObject(item.getSpec();/將 spec 字段中的 json5.商品刪除-移除 Solr 索引庫5.1 需求分析通過消息中間件實現(xiàn)在商品刪除時也

36、同時移除索引庫的功能。5.2 消息生產(chǎn)者（運營商）5.2.1 配置文件修改 pinyougou-manager-web 工程的 spring-activemq.xml，添加 bean 配置字符串轉換為 mapitem.setSpecMap(specMap);/給帶注解的字段賦值itemSearchService.importList(list);/導入 System.out.prln(成功導入到索引庫); catch (Exception e) e.prStackTrace();5.2.2 代碼實現(xiàn)修改GoodsController.javaAutowiredprivate Destinati

37、on queueSolrDeleteDestination;/用戶在索引庫中刪除/*批量刪除param idsreturn*/RequestMap(/delete)public Resuelete(final Long ids)try goodsService.delete(ids); jmsTemplate.send(queueSolrDeleteDestination, new MessageCreator() Overridepublic Message createMessage(Sesses) throws JMSException return ses.createObjectMe

38、ssage(ids););return new Result(true, 刪除成功); catch (Exception e) e.prStackTrace();5.3 消息消費者（搜索服務）5.3.1 配置文件修改 pinyougou-search-service 的 applicationContext-activemq-consumer.xml5.3.2 代碼實現(xiàn)com.pinyougou.search.service.impl 包下創(chuàng)建類 ItemDeleistener/*：用于刪除索引庫中author Administratorreturn new Result(false, 刪除失

39、敗);*/ Componentpublic class ItemDeleistener implements MessageListenerAutowiredprivate ItemSearchService itemSearchService;Overridepublic void onMessage(Message message) try ObjectMessage objectMessage= (ObjectMessage)message; Long goodsIds = (Long) objectMessage.getObject();System.out.prln(ItemDele

40、istener接收到消息.+goodsIds);itemSearchService.deleteByGoodsIds(Arrays.asList(goodsIds); System.out.prln(成功刪除索引庫中的); catch (Exception e) e.prStackTrace();6.商品審核-執(zhí)行網(wǎng)頁靜態(tài)化6.1 需求分析運用消息中間件 activeMQ 實現(xiàn)運營商與網(wǎng)頁生成服務的零耦合。運營商執(zhí)行商品審核后，向 activeMQ 發(fā)送消息（商品 ID），網(wǎng)頁生成服務從 activeMQ 接收到消息后執(zhí)行網(wǎng)頁生成操作。6.2 消息生產(chǎn)者（運營商）6.2.1 解除耦合修改 pi

41、nyougou-manager-web，移除網(wǎng)頁生成服務接口依賴：GoodsController.java 中刪除調用網(wǎng)頁生成服務接口的相關代碼/private ItemPageService itemPageService;com.pinyougoupinyougouerface0.0.3-SNAPSHOT6.2.2 準備工作修改配置文件spring-activemq.xml，添加配置6.2.3 代碼實現(xiàn)修改 pinyougou-manager-web 的GoodsController.javapublic Result updateSus(Long ids,String sus)try i

42、f(sus.equals(1)/審核通過/靜態(tài)頁生成for(final Long goodsId:ids) jmsTemplate.send(topicPageDestination, new MessageCreator() /靜態(tài)頁生成/for(Long goodsId:ids)/itemPageService.genItemHtml(goodsId);/6.3 消息消費者（頁面生成服務）6.3.1 解除 dubbox 依賴（1）修改工程 pinyougouservice ，刪除 dubbox 相關依賴com.alibabadubboOverridepublic Message creat

43、eMessage(Sesses) throwsJMSException return ses.createTextMessage(goodsId+);); catch (Exception e) （2）修改 applicationContext-service.xml，刪除 dubbox 相關配置（3）修改ItemPageServiceImpl 類的Service 注解 為包下的Service 注解.springframework.stereotype.Service6.3.2 準備工作（1）修改 applicationContext-service.xml，添加配置（2）pom.xml 中引

44、入 activeMQ 客戶端的依賴.apache.activemqactivemq-cnt5.13.4.apache.zookeeperzookeepercom.sgroschupfzkcnt（3）添加 spring 配置文件 applicationContext-jms-consumer.xml6.3.3 代碼編寫創(chuàng)建消息類PageListenerComponentpublic class PageListener implements MessageListener Autowiredprivate ItemPageService itemPageService;Overridepubli

45、c void onMessage(Message message) TextMessage textMessage= (TextMessage)message; try String text = textMessage.getText(); System.out.prln(接收到消息：+text);b = itemPageService.genItemHtml(Long.parseLong(text); catch (Exception e) e.prStackTrace();7.商品刪除-刪除商品詳細頁7.1 需求分析執(zhí)行商品刪除后，同時刪除每個服務器上的商品詳細頁7.2 消息生產(chǎn)者（運營商）7.2.1 配置文件修改 spring-activemq.xml，添加配置7.2.2 代碼實現(xiàn)修改GoodsController.javaAutowire