這篇文章主要介紹“怎么正確使用RabbitMQ異步編程”，在日常操作中，相信很多人在怎么正確使用RabbitMQ異步編程問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”怎么正確使用RabbitMQ異步編程”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！
成都創(chuàng)新互聯(lián)公司公司2013年成立，先為興城等服務(wù)建站，興城等地企業(yè)，進(jìn)行企業(yè)商務(wù)咨詢服務(wù)。為興城企業(yè)網(wǎng)站制作PC+手機(jī)+微官網(wǎng)三網(wǎng)同步一站式服務(wù)解決您的所有建站問題。
1 適用場景
1.1 服務(wù)于主流程的分支流程
在注冊流程中，數(shù)據(jù)寫DB是主流程，但注冊后給用戶發(fā)優(yōu)惠券或歡迎短信是分支流程，時(shí)效性也不強(qiáng)。
1.2 用戶無需實(shí)時(shí)看到結(jié)果
比如外賣下單后的配貨、送貨流程完全可異步處理，每個(gè)階段處理完成后，再給用戶發(fā)推送或短信讓用戶知曉即可。
1.3 MQ
任務(wù)的緩沖的分發(fā)，流量削峰、服務(wù)解耦和消息廣播。
當(dāng)然了異步處理不僅僅是通過 MQ 來實(shí)現(xiàn)，還有其他方式
比如開新線程執(zhí)行，返回 Future
還有各種異步框架，比如 Vertx，它是通過 callback 的方式實(shí)現(xiàn)
2 異步處理之坑
2.1 異步處理需做消息補(bǔ)償以閉環(huán)
RabbitMQ雖可將消息落地磁盤，即使MQ異常消息數(shù)據(jù)也不會(huì)丟失，但異步流程在消息發(fā)送、傳輸、處理等環(huán)節(jié)，都可能發(fā)生消息丟失。所有MQ都無法確保百分百可用，業(yè)務(wù)設(shè)計(jì)都需考慮不可用時(shí)異步流程將如何繼續(xù)。
因此，對于異步處理流程，必須考慮補(bǔ)償或建立主備雙活流程。
2.1.1 案例
用戶注冊后異步發(fā)送歡迎消息。
用戶注冊落DB為同步流程
會(huì)員服務(wù)收到消息后發(fā)送歡迎消息為異步流程
藍(lán)線
MQ異步處理(主線)，消息可能丟失(虛線代表異步調(diào)用)
綠線
補(bǔ)償Job定期消息補(bǔ)償(備線)，以補(bǔ)償主線丟失的消息
考慮極端的MQ中間件失效場景
要求備線的處理吞吐能力達(dá)到主線性能
代碼示例
UserController 注冊+發(fā)送異步消息。注冊方法，一次性注冊10個(gè)用戶，用戶注冊消息不能發(fā)送出去的概率為50%。
MemberService 會(huì)員服務(wù)監(jiān)聽用戶注冊成功的消息，并發(fā)送歡迎短信。使用ConcurrentHashMap存放那些發(fā)過短信的用戶ID實(shí)現(xiàn)冪等，避免相同的用戶補(bǔ)償時(shí)重復(fù)發(fā)短信
對于MQ消費(fèi)程序，處理邏輯須考慮去重(支持冪等)：
MQ消息可能會(huì)因中間件本身配置錯(cuò)誤、穩(wěn)定性等原因出現(xiàn)重復(fù)
自動(dòng)補(bǔ)償重復(fù)
比如本例，同一消息可能既走M(jìn)Q也走補(bǔ)償，肯定會(huì)出現(xiàn)重復(fù)，而且考慮到高內(nèi)聚，補(bǔ)償Job本身不會(huì)做去重
人工補(bǔ)償重復(fù)
出現(xiàn)消息堆積時(shí)，異步處理流程必然延遲。若提供補(bǔ)償功能，則在處理遇到延遲時(shí)，很可能會(huì)先人工補(bǔ)償，過段時(shí)間后處理程序又收到消息了，重復(fù)處理。
有次MQ故障，MQ中堆積了幾十萬條發(fā)放資金消息，導(dǎo)致業(yè)務(wù)無法及時(shí)處理，運(yùn)營以為程序出錯(cuò)，就先通過后臺進(jìn)行人工處理，結(jié)果MQ系統(tǒng)恢復(fù)后消息又被重復(fù)處理一次，造成大量資金重復(fù)發(fā)放。
異步處理須考慮消息重復(fù)可能性，因此處理邏輯須實(shí)現(xiàn)冪等，防止重復(fù)處理。
接著定義補(bǔ)償Job即備線操作。
定時(shí)任務(wù)，5秒做一次補(bǔ)償，因Job并不知道哪些用戶注冊的消息可能丟失，所以是全量補(bǔ)償。
補(bǔ)償邏輯
每5秒補(bǔ)償一次，按順序一次補(bǔ)償5個(gè)用戶，下一次補(bǔ)償操作從上一次補(bǔ)償?shù)淖詈笠粋€(gè)用戶ID開始
補(bǔ)償任務(wù)提交到線程池以“異步”處理，提高處理能力
為實(shí)現(xiàn)高內(nèi)聚，主線和備線處理消息，最好使用同一方法。本案例的MemberService監(jiān)聽到MQ消息和CompensationJob補(bǔ)償，調(diào)用的都是welcome。
這里的補(bǔ)償邏輯簡單僅為 demo，實(shí)際生產(chǎn)代碼須：
考慮配置補(bǔ)償?shù)念l次、每次處理數(shù)量，以及補(bǔ)償線程池大小等參數(shù)為合適值，以滿足補(bǔ)償?shù)耐掏铝?/p>
考慮備線補(bǔ)償數(shù)據(jù)進(jìn)行適當(dāng)延遲
比如，對注冊時(shí)間在30s前的用戶再進(jìn)行補(bǔ)償，以方便和主線MQ實(shí)時(shí)流程錯(cuò)開，避免沖突
諸如當(dāng)前補(bǔ)償?shù)侥膫€(gè)用戶的offset數(shù)據(jù)，需要落地DB
補(bǔ)償Job本身須高可用，可使用類似xxl-job或ElasticJob等任務(wù)系統(tǒng)。
運(yùn)行程序，執(zhí)行注冊方法注冊10個(gè)用戶，查看日志
可見
共10個(gè)用戶，MQ發(fā)送成功的用戶有四個(gè)：1、5、7、8
補(bǔ)償任務(wù)第一次運(yùn)行，補(bǔ)償了用戶2、3、4，第二次運(yùn)行補(bǔ)償了用戶6、9，第三次運(yùn)行補(bǔ)充了用戶10
消息補(bǔ)償閉環(huán)的最高標(biāo)準(zhǔn)
能夠達(dá)到補(bǔ)償全量數(shù)據(jù)的吞吐量。即若補(bǔ)償備線足夠完善，即使直接停機(jī)MQ，雖會(huì)稍微影響處理及時(shí)性，但至少確保流程都能正常執(zhí)行。
小結(jié)
實(shí)際開發(fā)要考慮異步流程丟消息或處理中斷場景。
異步流程需有備線以補(bǔ)償，比如這里的全量補(bǔ)償方式，即便異步流程徹底失效，通過補(bǔ)償也能讓業(yè)務(wù)繼續(xù)進(jìn)行。
2.2 RabbitMQ廣播、工作隊(duì)列模式坑
消息模式是廣播 Or 工作隊(duì)列
消息廣播
同一消息，不同消費(fèi)者都能分別消費(fèi)
隊(duì)列模式
不同消費(fèi)者共享消費(fèi)同一個(gè)隊(duì)列的數(shù)據(jù)，相同消息只能被某一個(gè)消費(fèi)者消費(fèi)一次。
比如同一用戶的注冊消息
會(huì)員服務(wù)需監(jiān)聽以發(fā)送歡迎短信
營銷服務(wù)需監(jiān)聽以發(fā)送新用戶小禮物
但會(huì)員、營銷服務(wù)都可能有多實(shí)例，業(yè)務(wù)需求同一用戶的消息，可同時(shí)廣播給不同的服務(wù)(廣播模式)，但對同一服務(wù)的不同實(shí)例(比如會(huì)員服務(wù)1和會(huì)員服務(wù)2)，不管哪個(gè)實(shí)例來處理，處理一次即可(工作隊(duì)列模式)：
實(shí)現(xiàn)代碼時(shí)務(wù)必確認(rèn)MQ系統(tǒng)的機(jī)制，確保消息的路由按期望。
RocketMQ實(shí)現(xiàn)類似功能比較簡單直白：若消費(fèi)者屬于一個(gè)組，那么消息只會(huì)由同組的一個(gè)消費(fèi)者消費(fèi);若消費(fèi)者屬不同組，每個(gè)組都能消費(fèi)一遍消息。
而RabbitMQ的消息路由模式采用隊(duì)列+交換器，隊(duì)列是消息載體，交換器決定消息路由到隊(duì)列的方式。
step1:會(huì)員服務(wù)-監(jiān)聽用戶服務(wù)發(fā)出的新用戶注冊消息
若啟動(dòng)倆會(huì)員服務(wù)，那么同一用戶的注冊消息應(yīng)只能被其中一個(gè)實(shí)例消費(fèi)。
分別實(shí)現(xiàn)RabbitMQ隊(duì)列、交換器、綁定三件套。
隊(duì)列使用匿名隊(duì)列
交換器使用DirectExchange，交換器綁定到匿名隊(duì)列的路由Key是空字符串
收到消息之后，打印所在實(shí)例使用的端口。
消息發(fā)布者、消費(fèi)者、以及MQ的配置
使用12345和45678兩個(gè)端口啟動(dòng)倆程序?qū)嵗?，發(fā)條消息，輸出的日志，顯示同一會(huì)員服務(wù)兩個(gè)實(shí)例都收到了消息：
所以問題在于不明
RabbitMQ直接交換器和隊(duì)列的綁定關(guān)系
RabbitMQ的直接交換器根據(jù)routingKey路由消息。而程序每次啟動(dòng)都會(huì)創(chuàng)建匿名(隨機(jī)命名)隊(duì)列，所以每個(gè)會(huì)員服務(wù)實(shí)例都對應(yīng)獨(dú)立的隊(duì)列，以空routingKey綁定到直接交換器。
用戶服務(wù)發(fā)消息時(shí)也設(shè)置了空routingKey，所以直接交換器收到消息后，發(fā)現(xiàn)匹配倆隊(duì)列，于是都轉(zhuǎn)發(fā)消息
修復(fù)
對會(huì)員服務(wù)不要使用匿名隊(duì)列，而使用同一隊(duì)列。
將上面代碼中的匿名隊(duì)列換做普通隊(duì)列：
private static final String QUEUE = "newuserQueue";@Beanpublic Queue queue() { return new Queue(QUEUE);}
這樣對同一消息，倆實(shí)例中只有一個(gè)實(shí)例可收到，不同消息被輪詢發(fā)給不同實(shí)例。
現(xiàn)在的交換器和隊(duì)列關(guān)系
step2:用戶服務(wù)-廣播消息給會(huì)員、營銷服務(wù)
期望會(huì)員、營銷服務(wù)都能收到廣播消息，但會(huì)員/營銷服務(wù)中的每個(gè)實(shí)例只需收到一次消息。
聲明一個(gè)隊(duì)列和一個(gè)FanoutExchange，然后模擬倆用戶服務(wù)和倆營銷服務(wù)：
注冊四個(gè)用戶。日志發(fā)現(xiàn)一條用戶注冊的消息，要么被會(huì)員服務(wù)收到，要么被營銷服務(wù)收到，這不是廣播?？墒褂玫拿髅魇荈anoutExchange，為什么沒起效呢?
因?yàn)閺V播交換器會(huì)忽略routingKey，廣播消息到所有綁定的隊(duì)列。該案例的倆會(huì)員服務(wù)和兩個(gè)營銷服務(wù)都綁定了同一隊(duì)列，所以四服務(wù)只能收到一次消息：
修復(fù)
拆分隊(duì)列，會(huì)員和營銷兩組服務(wù)分別使用一條獨(dú)立隊(duì)列綁定到廣播交換器
現(xiàn)在的交換器和隊(duì)列結(jié)構(gòu)
從日志輸出可以驗(yàn)證，對每條MQ消息，會(huì)員服務(wù)和營銷服務(wù)分別都會(huì)收到一次，一條消息廣播到兩個(gè)服務(wù)同時(shí)，在每一個(gè)服務(wù)的兩個(gè)實(shí)例中通過輪詢接收：
異步的消息路由模式一旦配置出錯(cuò)，輕則可能導(dǎo)致消息重復(fù)處理，重則可能導(dǎo)致重要的服務(wù)無法接收到消息，最終造成業(yè)務(wù)邏輯錯(cuò)誤。
小結(jié)
微服務(wù)場景下不同服務(wù)多個(gè)實(shí)例監(jiān)聽消息的情況，一般不同服務(wù)需要同時(shí)收到相同的消息，而相同服務(wù)的多個(gè)實(shí)例只需要輪詢接收消息。我們需要確認(rèn)MQ的消息路由配置是否滿足需求，以避免消息重復(fù)或漏發(fā)問題。
2.3 死信堵塞MQ之坑
始終無法處理的死信消息，可能會(huì)引發(fā)堵塞MQ。
若線程池的任務(wù)隊(duì)列無上限，最終可能導(dǎo)致OOM，類似的MQ也要注意任務(wù)堆積問題。對于突發(fā)流量引起的MQ堆積，問題并不大，適當(dāng)調(diào)整消費(fèi)者的消費(fèi)能力應(yīng)該就可以解決。但在很多時(shí)候，消息隊(duì)列的堆積堵塞，是因?yàn)橛写罅渴冀K無法處理的消息。
2.3.1 案例
用戶服務(wù)在用戶注冊后發(fā)出一條消息，會(huì)員服務(wù)監(jiān)聽到消息后給用戶派發(fā)優(yōu)惠券，但因用戶并沒有保存成功，會(huì)員服務(wù)處理消息始終失敗，消息重新進(jìn)入隊(duì)列，然后還是處理失敗。這種在MQ中回蕩的同一條消息，就是死信。
隨著MQ被越來越多的死信填滿，消費(fèi)者需花費(fèi)大量時(shí)間反復(fù)處理死信，導(dǎo)致正常消息的消費(fèi)受阻，最終MQ可能因數(shù)據(jù)量過大而崩潰。
定義一個(gè)隊(duì)列、一個(gè)直接交換器，然后把隊(duì)列綁定到交換器
sendMessage發(fā)送消息到MQ，訪問一次提交一條消息，使用自增標(biāo)識作為消息內(nèi)容
收到消息后，直接NPE，模擬處理出錯(cuò)
調(diào)用sendMessage接口發(fā)送兩條消息，然后來到RabbitMQ管理臺，可以看到這兩條消息始終在隊(duì)列，不斷被重新投遞，導(dǎo)致重新投遞QPS達(dá)到1063。
在日志中也可看到大量異常信息。
修復(fù)方案
解決死信無限重復(fù)進(jìn)入隊(duì)列最簡單方案
程序處理出錯(cuò)時(shí)，直接拋AmqpRejectAndDontRequeueException，避免消息重新進(jìn)入隊(duì)列
重慶分公司
重慶分公司
throw new AmqpRejectAndDontRequeueException("error");但更希望對同一消息，能夠先進(jìn)行幾次重試，解決因?yàn)榫W(wǎng)絡(luò)問題導(dǎo)致的偶發(fā)消息處理失敗，若依舊失敗，再把消息投遞到專門設(shè)置的DLX。對于來自DLX的數(shù)據(jù)，可能只是記錄日志發(fā)送報(bào)警，即使出現(xiàn)異常也不會(huì)再重復(fù)投遞。
邏輯如下
針對該問題，我們來看
Spring AMQP的簡便解決方案
鴻蒙官方戰(zhàn)略合作共建——HarmonyOS技術(shù)社區(qū)
定義死信交換器、死信隊(duì)列。其實(shí)都是普通交換器和隊(duì)列，只不過專門用于處理死信消息
通過RetryInterceptorBuilder構(gòu)建一個(gè)RetryOperationsInterceptor以處理失敗時(shí)候的重試。策略是最多嘗試5次(重試4次);并且采取指數(shù)退避重試，首次重試延遲1秒，第二次2秒，以此類推，最大延遲是10秒;如果第4次重試還是失敗，則使用RepublishMessageRecoverer把消息重新投入一個(gè)DLX
定義死信隊(duì)列的處理程序。本案例只記錄日志
代碼
執(zhí)行程序，發(fā)送兩條消息，查看日志：
msg2的4次重試間隔分別是1秒、2秒、4秒、8秒，再加上首次的失敗，所以最大嘗試次數(shù)是5
4次重試后，RepublishMessageRecoverer把消息發(fā)往DLX
死信處理程序輸出了got dead message msg2。
雖然幾乎同時(shí)發(fā)倆消息，但msg2在msg1四次重試全部結(jié)束后才開始處理，因?yàn)槟J(rèn)SimpleMessageListenerContainer只有一個(gè)消費(fèi)線程。可通過增加消費(fèi)線程避免性能問題：
直接設(shè)置concurrentConsumers參數(shù)為10，來增加到10個(gè)工作線程
也可設(shè)置maxConcurrentConsumers參數(shù)，讓SimpleMessageListenerContainer動(dòng)態(tài)調(diào)整消費(fèi)者線程數(shù)。
到此，關(guān)于“怎么正確使用RabbitMQ異步編程”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！
