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