Java應(yīng)用開(kāi)發(fā)（中、高級(jí)）課件 任務(wù)3.5 完成電商優(yōu)惠秒殺活動(dòng)熱插拔模塊實(shí)現(xiàn)

任務(wù)5.完成電商優(yōu)惠秒殺活動(dòng)熱插拔模塊實(shí)現(xiàn)任務(wù)描述前面的任務(wù)已經(jīng)完成了整個(gè)項(xiàng)目的核心功能開(kāi)發(fā)，本任務(wù)增加Rocketmq實(shí)現(xiàn)電商優(yōu)惠秒殺活動(dòng)熱插拔模塊。知識(shí)準(zhǔn)備?MQ背景&選型介紹：消息隊(duì)列作為高并發(fā)系統(tǒng)的核心組件之一，能夠幫助業(yè)務(wù)系統(tǒng)解構(gòu)提升開(kāi)發(fā)效率和系統(tǒng)穩(wěn)定性。主要具有以下優(yōu)勢(shì)：·削峰填谷（主要解決瞬時(shí)寫(xiě)壓力大于應(yīng)用服務(wù)能力導(dǎo)致消息丟失、系統(tǒng)奔潰等問(wèn)題）·系統(tǒng)解耦（解決不同重要程度、不同能力級(jí)別系統(tǒng)之間依賴導(dǎo)致一死全死）·提升性能（當(dāng)存在一對(duì)多調(diào)用時(shí)，可以發(fā)一條消息給消息系統(tǒng)，讓消息系統(tǒng)通知相關(guān)系統(tǒng)）·蓄流壓測(cè)（線上有些鏈路不好壓測(cè)，可以通過(guò)堆積一定量消息再放開(kāi)來(lái)壓測(cè)）目前主流的MQ主要是Rocketmq、kafka、Rabbitmq，Rocketmq相比于Rabbitmq、kafka具有主要優(yōu)勢(shì)特性有：·支持事務(wù)型消息（消息發(fā)送和DB操作保持兩方的最終一致性，rabbitmq和kafka不支持）?！ぶС纸Y(jié)合rocketmq的多個(gè)系統(tǒng)之間數(shù)據(jù)最終一致性（多方事務(wù)，二方事務(wù)是前提）。·支持18個(gè)級(jí)別的延遲消息（rabbitmq和kafka不支持）。·支持指定次數(shù)和時(shí)間間隔的失敗消息重發(fā)（kafka不支持，rabbitmq需要手動(dòng)確認(rèn)）?！ぶС謈onsumer端tag過(guò)濾，減少不必要的網(wǎng)絡(luò)傳輸（rabbitmq和kafka不支持）?！ぶС种貜?fù)消費(fèi)（rabbitmq不支持，kafka支持）。?RocketMQ介紹：RocketMQ是由阿里用java語(yǔ)言開(kāi)發(fā)的一款高性能、高吞吐量的分布式消息中間件，于2017年正式捐贈(zèng)Apache基金會(huì)并成為頂級(jí)開(kāi)源項(xiàng)目。1）事務(wù)基本概念①HalfMessage：也叫PrepareMessage，翻譯為“半消息”或“準(zhǔn)備消息”，指的是暫時(shí)無(wú)法投遞的消息，即消息成功發(fā)送到MQ服務(wù)器，暫時(shí)還不能給消費(fèi)者進(jìn)行消費(fèi)，只有當(dāng)服務(wù)器接收到生產(chǎn)者傳來(lái)的二次確認(rèn)時(shí)，才能被消費(fèi)者消費(fèi)②MessageStatusCheck：消息狀態(tài)回查。網(wǎng)絡(luò)斷開(kāi)連接或生產(chǎn)者應(yīng)用程序重新啟動(dòng)可能會(huì)丟失對(duì)事務(wù)性消息的第二次確認(rèn)，當(dāng)MQ服務(wù)器發(fā)現(xiàn)某條消息長(zhǎng)時(shí)間保持半消息狀態(tài)時(shí)，它會(huì)向消息生產(chǎn)者發(fā)送一個(gè)請(qǐng)求，去檢查消息的最終狀態(tài)（“提交”或“回滾”）2）RocketMQ執(zhí)行流程執(zhí)行流程如圖4-15所示：①支持重復(fù)消費(fèi)（rabbitmq不支持，kafka支持）。②生產(chǎn)者發(fā)送半消息到MQServer，暫時(shí)不能投遞，不會(huì)被消費(fèi)③半消息發(fā)送成功后，生產(chǎn)者這邊執(zhí)行本地事務(wù)④生產(chǎn)者根據(jù)本地事務(wù)執(zhí)行結(jié)果，向MQServer發(fā)送commit或rollback消息進(jìn)行二次確認(rèn)⑤如果MQServer接收到的commit，則將半消息標(biāo)記為可投遞狀態(tài)，此時(shí)消費(fèi)者就能進(jìn)行消費(fèi)；如果收到的是rollback，則將半消息直接丟棄，不會(huì)進(jìn)行消費(fèi)⑥如果MQServer未收到二次確認(rèn)消息，MQServer則會(huì)定時(shí)（默認(rèn)1分鐘）向生產(chǎn)者發(fā)送回查消息，檢查本地事務(wù)狀態(tài)，然后生產(chǎn)者根據(jù)本地事務(wù)回查結(jié)果再次向MQServer發(fā)送commit或rollback消息。圖4-15RocketMQ執(zhí)行流程圖?JMeter介紹：JMeter是Apache組織開(kāi)發(fā)的基于Java的壓力測(cè)試工具。用于對(duì)軟件做壓力測(cè)試，它最初被設(shè)計(jì)用于Web應(yīng)用測(cè)試，但后來(lái)擴(kuò)展到其他測(cè)試領(lǐng)域。它可以用于測(cè)試靜態(tài)和動(dòng)態(tài)資源，例如靜態(tài)文件、Java小服務(wù)程序、CGI腳本、Java對(duì)象、數(shù)據(jù)庫(kù)、FTP服務(wù)器，等等。JMeter可以用于對(duì)服務(wù)器、網(wǎng)絡(luò)或?qū)ο竽M巨大的負(fù)載，來(lái)自不同壓力類(lèi)別下測(cè)試它們的強(qiáng)度和分析整體性能。另外，JMeter能夠?qū)?yīng)用程序做功能/回歸測(cè)試，通過(guò)創(chuàng)建帶有斷言的腳本來(lái)驗(yàn)證你的程序返回了你期望的結(jié)果。為了最大限度的靈活性，JMeter允許使用正則表達(dá)式創(chuàng)建斷言。任務(wù)實(shí)施步驟1：搭建RocketMQ開(kāi)發(fā)環(huán)境(1)下載RocketMQ環(huán)境要求：?64bitOS,Linux/Unix/Macisrecommended;?64bitJDK1.8+;?Maven3.2.x;?Git;?4g+freediskforBrokerserver下載地址：/release_notes/release-notes-4.7.0/。下載并解壓：rocketmq-all-4.7.0-bin-release.zip。(2)配置環(huán)境變量

選中“此電腦”單擊鼠標(biāo)右鍵選擇“屬性->高級(jí)系統(tǒng)設(shè)置->高級(jí)->環(huán)境變量”，在系統(tǒng)變量中增加下列環(huán)境變量，詳細(xì)變量見(jiàn)表4-8。變量名變量值操作及描述ROCKETMQ_HOMEF:\rocketmq-all-4.7.0-bin-release新建，變量值為解壓縮地址。Path%ROCKETMQ_HOME%\bin;編輯，把變量值插入到原值前面即可。表4-8任務(wù)5RocketMQ環(huán)境變量如圖4-16所示：圖4-16設(shè)置RocketMQ環(huán)境變量(3)修改默認(rèn)配置本地開(kāi)發(fā)調(diào)試的話，不需要使用太高的配置.而默認(rèn)的配置占用的內(nèi)存太高，往往導(dǎo)致啟動(dòng)失敗，所以需要修改F:\rocketmq-all-4.7.0-bin-release\bin目錄下的默認(rèn)配置把啟動(dòng)內(nèi)存調(diào)?。簉unserver.cmdset"JAVA_OPT=%JAVA_OPT%-server-Xms2g-Xmx2g-Xmn1g-XX:MetaspaceSize=128m-XX:MaxMetaspaceSize=320m"修改為：set"JAVA_OPT=%JAVA_OPT%-server-Xms256m-Xmx256m-Xmn128m-XX:MetaspaceSize=128m-XX:MaxMetaspaceSize=320m"runbroker.cmdsetCLASSPATH=.;%BASE_DIR%conf;%CLASSPATH%setCLASSPATH=.;%BASE_DIR%conf;"%CLASSPATH%"set"JAVA_OPT=%JAVA_OPT%-server-Xms2g-Xmx2g-Xmn1g"set"JAVA_OPT=%JAVA_OPT%-server-Xms256m-Xmx256m-Xmn128m"修改為：修改為：(4)啟動(dòng)NameServer進(jìn)入bin目錄，執(zhí)行cmd命令：startmqnamesrv.cmd,執(zhí)行成功后，如圖4-17所示：圖4-17啟動(dòng)RocketMQNameServer(5)啟動(dòng)Broker繼續(xù)執(zhí)行命令：startmqbroker.cmd-nlocalhost:9876,執(zhí)行成功后，如圖4-18所示：圖4-18啟動(dòng)RocketMQBroker完成后，各個(gè)cmd命令窗口不要關(guān)閉，本地編寫(xiě)代碼進(jìn)行訪問(wèn)調(diào)試。步驟2：編寫(xiě)代碼實(shí)例(1)增加rocketmq依賴?打開(kāi)pom.xml,增加下列配置代碼：<!--rocketmq--><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.7.0</version></dependency>(1)在“com.chinasofti.ordersys.config”包中，新建JmsConfig類(lèi)，用于實(shí)現(xiàn)JMS配置。?代碼如下：packagecom.chinasofti.ordersys.config;

/***@authorandyzheng*@title:JmsConfig*@projectNameordersys*@description:TODO*@date2021/4/2623:44*/publicclassJmsConfig{

/***NameServer地址，因?yàn)槭羌翰渴鹚杂卸鄠€(gè)用分號(hào)隔開(kāi)*/publicstaticfinalStringNAME_SERVER="localhost:9876";/***主題名稱主題一般是服務(wù)器設(shè)置好而不能在代碼里去新建topic（如果沒(méi)有創(chuàng)建好，生產(chǎn)者往該主題發(fā)送消息會(huì)報(bào)找不到topic錯(cuò)誤）*/publicstaticfinalStringTOPIC="topic_family";}注意：NAME_SERVER必須與前面啟動(dòng)的broker的ip地址及端口號(hào)一致。(1)在“com.chinasofti.ordersys.config”包中，新建Producer類(lèi)，用于實(shí)現(xiàn)消息發(fā)送方。?消息發(fā)送方代碼如下：packagecom.chinasofti.ordersys.config;

importorg.apache.rocketmq.client.exception.MQClientException;importducer.DefaultMQProducer;importorg.springframework.stereotype.Component;

/***@authorandyzheng*@title:Producer*@projectNameordersys*@description:TODO*@date2021/4/2623:45*/@ComponentpublicclassProducer{privateStringproducerGroup="ORDER";privateDefaultMQProducerproducer;

publicProducer(){//示例生產(chǎn)者producer=newDefaultMQProducer(producerGroup);//不開(kāi)啟vip通道開(kāi)通口端口會(huì)減2producer.setVipChannelEnabled(false);//綁定nameserverproducer.setNamesrvAddr(JmsConfig.NAME_SERVER);start();}/***對(duì)象在使用之前必須要調(diào)用一次，只能初始化一次*/publicvoidstart(){try{ducer.start();}catch(MQClientExceptione){e.printStackTrace();}}

publicDefaultMQProducergetProducer(){returnducer;}/***一般在應(yīng)用上下文，使用上下文監(jiān)聽(tīng)器，進(jìn)行關(guān)閉*/publicvoidshutdown(){ducer.shutdown();}}(4)在“com.chinasofti.ordersys.config”包中，新建Consumer類(lèi)，用于實(shí)現(xiàn)消費(fèi)者。?

消費(fèi)者代碼如下：packagecom.chinasofti.ordersys.config;

importcom.chinasofti.ordersys.service.waiter.OrderService;importorg.apache.rocketmq.client.consumer.DefaultMQPushConsumer;importorg.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;importorg.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;importorg.apache.rocketmq.client.exception.MQClientException;importmon.consumer.ConsumeFromWhere;importmon.message.Message;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;

importjava.io.UnsupportedEncodingException;

importstaticjava.lang.Thread.sleep;/***@authorandyzheng*@title:Consumer*@projectNameordersys*@description:TODO*@date2021/4/2623:46*/@ComponentpublicclassConsumer{publicstaticfinalLoggerLOGGER=LoggerFactory.getLogger(Consumer.class);

/***消費(fèi)者實(shí)體對(duì)象*/privateDefaultMQPushConsumerconsumer;

@AutowiredprivateOrderServiceorderService;/***消費(fèi)者組*/publicstaticfinalStringCONSUMER_GROUP="ORDER";/***通過(guò)構(gòu)造函數(shù)實(shí)例化對(duì)象*/publicConsumer()throwsMQClientException{

consumer=newDefaultMQPushConsumer(CONSUMER_GROUP);consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);//消費(fèi)模式:一個(gè)新的訂閱組第一次啟動(dòng)從隊(duì)列的最后位置開(kāi)始消費(fèi)后續(xù)再啟動(dòng)接著上次消費(fèi)的進(jìn)度開(kāi)始消費(fèi)consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);//訂閱主題和標(biāo)簽（*代表所有標(biāo)簽)下信息consumer.subscribe(JmsConfig.TOPIC,"*");////注冊(cè)消費(fèi)的監(jiān)聽(tīng)并在此監(jiān)聽(tīng)中消費(fèi)信息，并返回消費(fèi)的狀態(tài)信息consumer.registerMessageListener((MessageListenerConcurrently)(msgs,context)->{//msgs中只收集同一個(gè)topic，同一個(gè)tag，并且key相同的message//會(huì)把不同的消息分別放置到不同的隊(duì)列中try{for(Messagemsg:msgs){

//消費(fèi)者獲取消息這里只輸出不做后面邏輯處理Stringbody=newString(msg.getBody(),"utf-8");LOGGER.info("Consumer-獲取消息-主題topic為={},消費(fèi)消息為={}",msg.getTopic(),body);if(body!=null){intorderId=Integer.parseInt(body);finalintSTATE=1;sleep(5000L);orderService.updateState(STATE,orderId);}}}catch(UnsupportedEncodingException|InterruptedExceptione){e.printStackTrace();returnConsumeConcurrentlyStatus.RECONSUME_LATER;}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;});

consumer.start();System.out.println("消費(fèi)者啟動(dòng)成功=======");}}(5)在“com.chinasofti.ordersys.api.waiter”包中，修改WaiterDishesController類(lèi)，實(shí)現(xiàn)秒殺功能。?增加引入類(lèi)如下：importmon.result.Results;importcom.chinasofti.ordersys.config.JmsConfig;importcom.chinasofti.ordersys.config.Producer;importducer.SendResult;importmon.message.Message;?增加代碼如下：@AutowiredprivateProducerproducer;

@PostMapping("/checkout")publicResults<?>checkout(HttpServletRequestrequest,@RequestBodyList<OrderDishesVO>orderDishes)throwsInterruptedException,RemotingException,MQClientException,MQBrokerException{LoginUserloginUser=tokenService.getLoginUser(request);if(loginUser==null){returnResults.failure();}UserInfouser=loginUser.getUser();intuserId=user.getUserId();OrderInfoorderInfo=orderService.saveOrder(userId);if(orderInfo==null||orderInfo.getOrderId()==null){returnResults.failure();}orderService.saveOrderDishes(orderInfo.getOrderId(),orderDishes);Stringmsg=orderInfo.getOrderId().toString();Messagemessage=newMessage(JmsConfig.TOPIC,"checkout",msg.getBytes());SendResultsendResult=producer.getProducer().send(message);returnResults.success();}步驟3：進(jìn)行高并發(fā)應(yīng)壓力測(cè)試(1)進(jìn)入官網(wǎng)下載并解壓縮Jmeter環(huán)境要求：?Java8+官網(wǎng)：/官網(wǎng)下載頁(yè)面：/download_jmeter.cgi下載頁(yè)面如圖4-19所示：圖4-19Jmeter下載頁(yè)面(2)進(jìn)到j(luò)meter安裝路徑，進(jìn)入bin文件夾，雙擊jmeter.bat啟動(dòng)，如圖4-20所示：圖4-20Jmeter界面(3)添加ThreadGroup，選擇"Edit->Add->Threads(Users)->ThreadGroup"，如圖4-21所示：圖4-21添加ThreadGroup(4)配置ThreadGroup并發(fā)數(shù)，60秒內(nèi)完成1000個(gè)請(qǐng)求線程，如圖4-22所示：4-22ThreadGroup配置界面(5)添加HttpRequest，選中ThreadGroup單擊鼠標(biāo)右鍵“Add->Samlter->HttpRequest“，如圖4-23所示：圖4-23添加HttpRequest(6)設(shè)置HttpRequest，如圖4-24所示：圖4-24HttpRequest配置界面HttpRequest設(shè)置明細(xì)表見(jiàn)表4-9。表4-9任務(wù)5HttpRequest設(shè)置明細(xì)表設(shè)置項(xiàng)輸入值描述Protocol[http]http請(qǐng)求協(xié)議。ServerNameorIPlocalhost請(qǐng)求服務(wù)器名稱或IP地址。PortNumber8080端口號(hào)。HTTPRequestPOST請(qǐng)求方式。Path/waiter/dishes/checkout請(qǐng)求地址。Browser-compatibleheaders√瀏覽器兼容的標(biāo)題。BodyData[{"dishesId":5,"num":1},{"dishesId":6,"num":1}]POST請(qǐng)求參數(shù)以JSON格式提交。當(dāng)前參數(shù)值表示：下5號(hào)菜、6號(hào)菜各1份。(7)添加HTTPHeaderManager，選中HttpRequest單擊鼠標(biāo)右鍵“Add->ConfigElement->HTTPHeaderManager“，如圖4-25所示：圖4-25添加HTTPHeaderManager(8)登錄服務(wù)員賬號(hào)后，在后臺(tái)控制臺(tái)獲取用戶token，如圖4-26所示：圖4-26后臺(tái)控制臺(tái)信息(9)設(shè)置HTTP