點(diǎn)擊上方“Java技術(shù)江湖”，選擇“設(shè)為星標(biāo)”

回復(fù)”666“獲取全網(wǎng)最熱的Java核心知識(shí)點(diǎn)整理

消息中間件是分布式系統(tǒng)常用的組件，無(wú)論是異步化、解耦、削峰等都有廣泛的應(yīng)用價(jià)值。我們通常會(huì)認(rèn)為，消息中間件是一個(gè)可靠的組件——這里所謂的可靠是指，只要我把消息成功投遞到了消息中間件，消息就不會(huì)丟失，即消息肯定會(huì)至少保證消息能被消費(fèi)者成功消費(fèi)一次，這是消息中間件最基本的特性之一，也就是我們常說(shuō)的“AT LEAST ONCE”，即消息至少會(huì)被“成功消費(fèi)一遍”。

舉個(gè)例子，一個(gè)消息M發(fā)送到了消息中間件，消息投遞到了消費(fèi)程序A，A接受到了消息，然后進(jìn)行消費(fèi)，但在消費(fèi)到一半的時(shí)候程序重啟了，這時(shí)候這個(gè)消息并沒有標(biāo)記為消費(fèi)成功，這個(gè)消息還會(huì)繼續(xù)投遞給這個(gè)消費(fèi)者，直到其消費(fèi)成功了，消息中間件才會(huì)停止投遞。

然而這種可靠的特性導(dǎo)致，消息可能被多次地投遞。舉個(gè)例子，還是剛剛這個(gè)例子，程序A接受到這個(gè)消息M并完成消費(fèi)邏輯之后，正想通知消息中間件“我已經(jīng)消費(fèi)成功了”的時(shí)候，程序就重啟了，那么對(duì)于消息中間件來(lái)說(shuō)，這個(gè)消息并沒有成功消費(fèi)過(guò)，所以他還會(huì)繼續(xù)投遞。這時(shí)候?qū)τ趹?yīng)用程序A來(lái)說(shuō)，看起來(lái)就是這個(gè)消息明明消費(fèi)成功了，但是消息中間件還在重復(fù)投遞。

這在RockectMQ的場(chǎng)景來(lái)看，就是同一個(gè)messageId的消息重復(fù)投遞下來(lái)了。

基于消息的投遞可靠（消息不丟）是優(yōu)先級(jí)更高的，所以消息不重的任務(wù)就會(huì)轉(zhuǎn)移到應(yīng)用程序自我實(shí)現(xiàn)，這也是為什么RocketMQ的文檔里強(qiáng)調(diào)的，消費(fèi)邏輯需要自我實(shí)現(xiàn)冪等。背后的邏輯其實(shí)就是：不丟和不重是矛盾的（在分布式場(chǎng)景下），但消息重復(fù)是有解決方案的，而消息丟失是很麻煩的。

簡(jiǎn)單的消息去重解決方案

1
2

insert into t_order values .....
update t_inv set count = count-1 where good_id = 'good123';

要實(shí)現(xiàn)消息的冪等，我們可能會(huì)采取這樣的方案：

1
2
3
4
5
6
7

select * from t_order where order_no = 'order123'
if(order  != null) {
    return ;//消息重復(fù)，直接返回
}

這對(duì)于很多情況下，的確能起到不錯(cuò)的效果，但是在并發(fā)場(chǎng)景下，還是會(huì)有問(wèn)題。

并發(fā)重復(fù)消息

假設(shè)這個(gè)消費(fèi)的所有代碼加起來(lái)需要1秒，有重復(fù)的消息在這1秒內(nèi)（假設(shè)100毫秒）內(nèi)到達(dá)（例如生產(chǎn)者快速重發(fā)，Broker重啟等），那么很可能，上面去重代碼里面會(huì)發(fā)現(xiàn)，數(shù)據(jù)依然是空的（因?yàn)樯弦粭l消息還沒消費(fèi)完，還沒成功更新訂單狀態(tài)），

那么就會(huì)穿透掉檢查的擋板，最后導(dǎo)致重復(fù)的消息消費(fèi)邏輯進(jìn)入到非冪等安全的業(yè)務(wù)代碼中，從而引發(fā)重復(fù)消費(fèi)的問(wèn)題（如主鍵沖突拋出異常、庫(kù)存被重復(fù)扣減而沒釋放等）

并發(fā)去重的解決方案之一

要解決上面并發(fā)場(chǎng)景下的消息冪等問(wèn)題，一個(gè)可取的方案是開啟事務(wù)把select 改成 select for update語(yǔ)句，把記錄進(jìn)行鎖定。

1
2
3
4

select * from t_order where order_no = 'THIS_ORDER_NO' for update  //開啟事務(wù)
if(order.status != null) {
    return ;//消息重復(fù)，直接返回
}

但這樣消費(fèi)的邏輯會(huì)因?yàn)橐肓耸聞?wù)包裹而導(dǎo)致整個(gè)消息消費(fèi)可能變長(zhǎng)，并發(fā)度下降。

當(dāng)然還有其他更高級(jí)的解決方案，例如更新訂單狀態(tài)采取樂(lè)觀鎖，更新失敗則消息重新消費(fèi)之類的。但這需要針對(duì)具體業(yè)務(wù)場(chǎng)景做更復(fù)雜和細(xì)致的代碼開發(fā)、庫(kù)表設(shè)計(jì)，不在本文討論的范圍。

但無(wú)論是select for update， 還是樂(lè)觀鎖這種解決方案，實(shí)際上都是基于業(yè)務(wù)表本身做去重，這無(wú)疑增加了業(yè)務(wù)開發(fā)的復(fù)雜度， 一個(gè)業(yè)務(wù)系統(tǒng)里面很大部分的請(qǐng)求處理都是依賴MQ的，如果每個(gè)消費(fèi)邏輯本身都需要基于業(yè)務(wù)本身而做去重/冪等的開發(fā)的話，這是繁瑣的工作量。本文希望探索出一個(gè)通用的消息冪等處理的方法，從而抽象出一定的工具類用以適用各個(gè)業(yè)務(wù)場(chǎng)景。

Exactly Once

在消息中間件里，有一個(gè)投遞語(yǔ)義的概念，而這個(gè)語(yǔ)義里有一個(gè)叫”Exactly Once”，即消息肯定會(huì)被成功消費(fèi)，并且只會(huì)被消費(fèi)一次。以下是阿里云里對(duì)Exactly Once的解釋：

Exactly-Once 是指發(fā)送到消息系統(tǒng)的消息只能被消費(fèi)端處理且僅處理一次，即使生產(chǎn)端重試消息發(fā)送導(dǎo)致某消息重復(fù)投遞，該消息在消費(fèi)端也只被消費(fèi)一次。

基于關(guān)系數(shù)據(jù)庫(kù)事務(wù)插入消息表

1

update t_order set status = 'SUCCESS' where order_no= 'order123';

要實(shí)現(xiàn)Exaclty Once即這個(gè)消息只被消費(fèi)一次（并且肯定要保證能消費(fèi)一次），我們可以這樣做：在這個(gè)數(shù)據(jù)庫(kù)中增加一個(gè)消息消費(fèi)記錄表，把消息插入到這個(gè)表，并且把原來(lái)的訂單更新和這個(gè)插入的動(dòng)作放到同一個(gè)事務(wù)中一起提交，就能保證消息只會(huì)被消費(fèi)一遍了。

1. 開啟事務(wù)
2. 插入消息表（處理好主鍵沖突的問(wèn)題）
3. 更新訂單表（原消費(fèi)邏輯）
4. 提交事務(wù)

1. 這時(shí)候如果消息消費(fèi)成功并且事務(wù)提交了，那么消息表就插入成功了，這時(shí)候就算RocketMQ還沒有收到消費(fèi)位點(diǎn)的更新再次投遞，也會(huì)插入消息失敗而視為已經(jīng)消費(fèi)過(guò)，后續(xù)就直接更新消費(fèi)位點(diǎn)了。這保證我們消費(fèi)代碼只會(huì)執(zhí)行一次。
2. 如果事務(wù)提交之前服務(wù)掛了（例如重啟），對(duì)于本地事務(wù)并沒有執(zhí)行所以訂單沒有更新，消息表也沒插入成功；而對(duì)于RocketMQ服務(wù)端來(lái)說(shuō)，消費(fèi)位點(diǎn)也沒更新，所以消息還會(huì)繼續(xù)投遞下來(lái)，投遞下來(lái)發(fā)現(xiàn)這個(gè)消息插入消息表也是成功的，所以可以繼續(xù)消費(fèi)。這保證了消息不丟失。

1. 消息的消費(fèi)邏輯必須是依賴于關(guān)系型數(shù)據(jù)庫(kù)事務(wù)。如果消費(fèi)的消費(fèi)過(guò)程中還涉及其他數(shù)據(jù)的修改，例如Redis這種不支持事務(wù)特性的數(shù)據(jù)源，則這些數(shù)據(jù)是不可回滾的。
2. 數(shù)據(jù)庫(kù)的數(shù)據(jù)必須是在一個(gè)庫(kù)，跨庫(kù)無(wú)法解決

更復(fù)雜的業(yè)務(wù)場(chǎng)景

1. 檢查庫(kù)存（RPC）
2. 鎖庫(kù)存（RPC）
3. 開啟事務(wù)，插入訂單表（MySQL）
4. 調(diào)用某些其他下游服務(wù)（RPC）
5. 更新訂單狀態(tài)
6. commit 事務(wù)（MySQL）

拆解消息執(zhí)行過(guò)程

1. 庫(kù)存系統(tǒng)消費(fèi)A：檢查庫(kù)存并做鎖庫(kù)存，發(fā)送消息B給訂單服務(wù)
2. 訂單系統(tǒng)消費(fèi)消息B：插入訂單表（MySQL），發(fā)送消息C給自己（下游系統(tǒng)）消費(fèi)
3. 下游系統(tǒng)消費(fèi)消息C：處理部分邏輯，發(fā)送消息D給訂單系統(tǒng)
4. 訂單系統(tǒng)消費(fèi)消息D：更新訂單狀態(tài)

更通用的解決方案

基于消息冪等表的非事務(wù)方案

1. 消息已經(jīng)消費(fèi)成功了，第二條消息將被直接冪等處理掉（消費(fèi)成功）。
2. 并發(fā)場(chǎng)景下的消息，依舊能滿足不會(huì)出現(xiàn)消息重復(fù)，即穿透冪等擋板的問(wèn)題。
3. 支持上游業(yè)務(wù)生產(chǎn)者重發(fā)的業(yè)務(wù)重復(fù)的消息冪等問(wèn)題。

此方案是否有消息丟失的風(fēng)險(xiǎn)？

更靈活的消息表存儲(chǔ)媒介

1. 性能上損耗更低
2. 上面我們講到的超時(shí)時(shí)間可以直接利用Redis本身的ttl實(shí)現(xiàn)

源碼：RocketMQDedupListener

1
2
3
4
5
6
7
8
9
10
11

        //利用Redis做冪等表
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP1");
        consumer.subscribe("TEST-TOPIC", "*");
        String appName = consumer.getConsumerGroup();// 大部分情況下可直接使用consumer group名
        StringRedisTemplate stringRedisTemplate = null;// 這里省略獲取StringRedisTemplate的過(guò)程
        DedupConfig dedupConfig = DedupConfig.enableDedupConsumeConfig(appName, stringRedisTemplate);
        DedupConcurrentListener messageListener = new SampleListener(dedupConfig);
        consumer.registerMessageListener(messageListener);
        consumer.start();

這種實(shí)現(xiàn)是否一勞永逸？

1. 檢查庫(kù)存（RPC）
2. 鎖庫(kù)存（RPC）
3. 開啟事務(wù)，插入訂單表（MySQL）
4. 調(diào)用某些其他下游服務(wù)（RPC）
5. 更新訂單狀態(tài)
6. commit 事務(wù)（MySQL）

本實(shí)現(xiàn)方式的價(jià)值？

一些其他的消息去重的建議

1. 消息消費(fèi)失敗做好回滾處理。如果消息消費(fèi)失敗本身是帶回滾機(jī)制的，那么消息重試自然就沒有副作用了。
2. 消費(fèi)者做好優(yōu)雅退出處理。這是為了盡可能避免消息消費(fèi)到一半程序退出導(dǎo)致的消息重試。
3. 一些無(wú)法做到冪等的操作，至少要做到終止消費(fèi)并告警。例如鎖庫(kù)存的操作，如果統(tǒng)一的業(yè)務(wù)流水鎖成功了一次庫(kù)存，再觸發(fā)鎖庫(kù)存，如果做不到冪等的處理，至少要做到消息消費(fèi)觸發(fā)異常（例如主鍵沖突導(dǎo)致消費(fèi)異常等）
4. 在#3做好的前提下，做好消息的消費(fèi)監(jiān)控，發(fā)現(xiàn)消息重試不斷失敗的時(shí)候，手動(dòng)做好#1的回滾，使得下次重試消費(fèi)成功。

轉(zhuǎn)自：薛定諤的風(fēng)口豬

鏈接：https://jaskey.github.io/blog/2020/06/08/rocketmq-message-dedup/

Java 項(xiàng)目構(gòu)建基礎(chǔ)：統(tǒng)一結(jié)果，統(tǒng)一異常，統(tǒng)一日志

妙用Java8中的“Function”消滅if...else！

Spring Boot 一個(gè)接口同時(shí)支持 form 表單、form-data、json 的優(yōu)雅寫法

關(guān)注公眾號(hào)【Java技術(shù)江湖】后回復(fù)“PDF”即可領(lǐng)取200+頁(yè)的《Java工程師面試指南》

強(qiáng)烈推薦，幾乎涵蓋所有Java工程師必知必會(huì)的知識(shí)點(diǎn)，不管是復(fù)習(xí)還是面試，都很實(shí)用。