點擊上方藍色字體，選擇“標星公眾號”

優(yōu)質文章，第一時間送達

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

select * from t_order where order_no = 'order123'

if(order  != null) {

    return ;//消息重復，直接返回

}

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

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

Exactly Once

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

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

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

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

1. 消息的消費邏輯必須是依賴于關系型數據庫事務。如果消費的消費過程中還涉及其他數據的修改，例如Redis這種不支持事務特性的數據源，則這些數據是不可回滾的。
2. 數據庫的數據必須是在一個庫，跨庫無法解決

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

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

基于消息冪等表的非事務方案

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

此方案是否有消息丟失的風險？

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

源碼：RocketMQDedupListener

        //利用Redis做冪等表
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP1");
        consumer.subscribe("TEST-TOPIC", "*");

        String appName = consumer.getConsumerGroup();// 大部分情況下可直接使用consumer group名
        StringRedisTemplate stringRedisTemplate = null;// 這里省略獲取StringRedisTemplate的過程
        DedupConfig dedupConfig = DedupConfig.enableDedupConsumeConfig(appName, stringRedisTemplate);
        DedupConcurrentListener messageListener = new SampleListener(dedupConfig);

        consumer.registerMessageListener(messageListener);
        consumer.start();

這種實現是否一勞永逸？

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

本實現方式的價值？

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

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

  作者 |  jaskey

來源 |  jaskey.github.io/blog/2020/06/08/rocketmq-message-dedup/