當(dāng)前位置：首頁 > 科技 > 軟件

實戰(zhàn)與原理：如何基于RocketMQ實現(xiàn)分布式事務(wù)？

來源：責(zé)編：時間：2024-01-26 08:59:37 237觀看

導(dǎo)讀使用事務(wù)消息在DailyMart系統(tǒng)中，用戶發(fā)起支付后，訂單系統(tǒng)需要調(diào)用庫存服務(wù)執(zhí)行庫存扣減邏輯。由于這是跨服務(wù)調(diào)用，因此會產(chǎn)生分布式事務(wù)。在這里，我們使用RocketMQ的事務(wù)消息來實現(xiàn)分布式事務(wù)。1、首先，在訂單服務(wù)的應(yīng)用服

使用事務(wù)消息

在DailyMart系統(tǒng)中，用戶發(fā)起支付后，訂單系統(tǒng)需要調(diào)用庫存服務(wù)執(zhí)行庫存扣減邏輯。

由于這是跨服務(wù)調(diào)用，因此會產(chǎn)生分布式事務(wù)。在這里，我們使用RocketMQ的事務(wù)消息來實現(xiàn)分布式事務(wù)。

1、首先，在訂單服務(wù)的應(yīng)用服務(wù)層處理支付邏輯，并調(diào)用RocketMQ發(fā)送事務(wù)消息：

@Overridepublic String payment(String orderSn) {    // todo 集成支付寶支付    // 支付流水號    String outOrderNo = IdUtils.get32UUID();    TradeOrder tradeOrder = Optional.ofNullable(tradeOrderService.getByOrderSn(orderSn)).orElseThrow(() -> new BusinessException("訂單編號不存在"));    // 如果訂單處于待支付狀態(tài)    if (Objects.equals(tradeOrder.getStatus(), OrderStatusEnum.WAITING_PAYMENT.getStatus())) {        OrderPaidEvent orderPaidEvent = new OrderPaidEvent(orderSn, outOrderNo);        TransactionSendResult sendResult = enhanceTemplate.sendTransaction("TRADE-ORDER", "ORDER-PAID");        if (SendStatus.SEND_OK == sendResult.getSendStatus() && sendResult.getLocalTransactionState() == LocalTransactionState.COMMIT_MESSAGE) {            return tradeOrder.getOrderSn();        } else {            throw new BusinessException("支付失敗...");        }    } else {        throw new BusinessException("訂單已支付，請勿重復(fù)提交...");    }}

2、在訂單服務(wù)的基礎(chǔ)設(shè)施層，創(chuàng)建一個類實現(xiàn) RocketMQLocalTransactionListener 接口：

該接口有兩個方法：

executeLocalTransaction：用于執(zhí)行本地事務(wù)。
checkLocalTransaction：在RocketMQ執(zhí)行消息回查時檢查本地事務(wù)執(zhí)行結(jié)果，用于確定消息提交還是回滾。

@Component@Slf4jpublic class OrderPaidTransactionConsumer implements RocketMQLocalTransactionListener {        @Resource    private TransactionTemplate transactionTemplate;    @Resource    private TradeOrderService tradeOrderService;         /**     * 執(zhí)行本地事務(wù)     * 將訂單狀態(tài)修改成已支付     */    @Override    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {                final OrderPaidEvent orderPaidEvent = JsonUtils.byte2Obj((byte[]) message.getPayload(), OrderPaidEvent.class);        try {            // 放到同一個本地事務(wù)中            this.transactionTemplate.executeWithoutResult(status -> {                String orderSn = orderPaidEvent.getOrderSn();                // 修改成待發(fā)貨                tradeOrderService.changeOrderStatus(orderSn, OrderStatusEnum.AWAITING_SHIPMENT);            });            return RocketMQLocalTransactionState.COMMIT;        } catch (Exception e) {            log.error("修改訂單狀態(tài)失敗", e);            // ROLLBACK 則回滾消息，rocketmq將廢棄這條消息            return RocketMQLocalTransactionState.ROLLBACK;            // 如果是UNKNOWN, 則觸發(fā)回查        }    }    /**     * 檢查本地事務(wù)執(zhí)行狀態(tài)     * 消息回查時，對于正在進(jìn)行中的事務(wù)不要返回Rollback或Commit結(jié)果，應(yīng)繼續(xù)保持Unknown的狀態(tài)。     */    @Override    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {        final OrderPaidEvent orderPaidEvent = JsonUtils.byte2Obj((byte[]) message.getPayload(), OrderPaidEvent.class);        String orderSn = orderPaidEvent.getOrderSn();        TradeOrder tradeOrder = tradeOrderService.getByOrderSn(orderSn);        // 如果已經(jīng)修改成待發(fā)貨說明本地事務(wù)執(zhí)行成功，此時消費端可以直接消費        if (Objects.equals(tradeOrder.getStatus(), OrderStatusEnum.AWAITING_SHIPMENT.getStatus())) {            return RocketMQLocalTransactionState.COMMIT;        } else {            // 這里查不到的時候返回 UNKNOWN在于，有可能事務(wù)還沒有提交，回查就開始了            return RocketMQLocalTransactionState.UNKNOWN;        }    }}

3、在庫存服務(wù)的基礎(chǔ)設(shè)施層，監(jiān)聽消息以執(zhí)行庫存扣減邏輯：

@Component@Slf4j@RocketMQMessageListener(consumerGroup = "dailymart_inventory_group", topic = "TRADE-ORDER", selectorExpression = "ORDER-PAID")public class InventoryDeductionConsumer extends EnhanceMessageHandler<OrderPaidEvent> implements RocketMQListener<OrderPaidEvent> {        @Resource    private InventoryDomainService inventoryDomainService;        @Override    public void onMessage(OrderPaidEvent orderPaidEvent) {        super.dispatchMessage(orderPaidEvent);    }        @Override    protected void handleMessage(OrderPaidEvent orderPaidEvent) throws Exception {        // 執(zhí)行庫存扣減邏輯        String orderSn = orderPaidEvent.getOrderSn();        inventoryDomainService.deductionInventory(orderSn);    }}

通過以上步驟，我們完成了RocketMQ事務(wù)消息的發(fā)送，利用事務(wù)消息的特性保證分布式事務(wù)的最終一致性。與普通消息相比，事務(wù)消息在處理時需要實現(xiàn) RocketMQLocalTransactionListener 接口，這是事務(wù)消息的核心。

介紹完事務(wù)消息的使用，接下來我們再來聊聊事務(wù)消息的原理。

事務(wù)消息的原理

首先，讓我們思考一下，如果不使用事務(wù)消息會有什么問題。

很容易想到的一個問題就是消息丟失。當(dāng)保存訂單后由于網(wǎng)絡(luò)問題導(dǎo)致消息丟失，如下圖所示：

圖片

在不使用RocketMQ的情況下，我們往往會通過本地消息表 + 補(bǔ)償重試的機(jī)制來保證消息一定會發(fā)送出去。其原理可以參考上篇文章 [Dailymart26：微服務(wù)中躲不過的坑 - 分布式事務(wù)]。

那RocketMQ是如何解決這個問題的呢？

1. 發(fā)送half消息，探測MQ是否正常

在基于RocketMQ的事務(wù)消息中，我們不是先執(zhí)行自身的訂單支付邏輯，而是先讓訂單系統(tǒng)發(fā)送一條 half消息到MQ去。這個half消息本質(zhì)上是一個訂單支付成功的消息，只不過此時庫存系統(tǒng)是看不見這個half消息的。然后，我們等待接收這個half消息寫入成功的響應(yīng)通知。

圖片

發(fā)送half消息的本質(zhì)其實是為了探測MQ是否仍然正常運(yùn)行。但問題來了，如上所述，消息會發(fā)生丟失，那么half消息丟失怎么辦呢？

2. half消息發(fā)送失敗

在發(fā)送half消息時，由于網(wǎng)絡(luò)原因或者M(jìn)Q直接掛了，就會導(dǎo)致half消息發(fā)送失敗。這個時候訂單系統(tǒng)需要執(zhí)行一系列的回滾操作。在我們的場景中，應(yīng)該執(zhí)行退款操作，將錢退還給用戶，并告知用戶交易失敗。

3. half消息成功，訂單系統(tǒng)執(zhí)行自己的業(yè)務(wù)邏輯

如果成功收到half消息的正常響應(yīng)，此時訂單系統(tǒng)應(yīng)該執(zhí)行自己的業(yè)務(wù)邏輯。在我們這個場景中，就是修改訂單數(shù)據(jù)庫狀態(tài)，將其修改為待發(fā)貨狀態(tài)。這部分邏輯就對應(yīng)上述代碼中的executeLocalTransaction()方法。

圖片

4. 訂單本地事務(wù)執(zhí)行失敗

如果訂單系統(tǒng)執(zhí)行本地事務(wù)失敗，則需要發(fā)送一個rollback請求給MQ，讓其刪除這條half消息。

圖片

5. 訂單本地事務(wù)執(zhí)行成功

如果訂單系統(tǒng)的本地事務(wù)執(zhí)行正常，此時需要發(fā)送一個commit請求給MQ，要求MQ對之前的half消息進(jìn)行commit操作，這樣庫存系統(tǒng)就可以消費這條消息了。

圖片

訂單創(chuàng)建消息處于half狀態(tài)時，庫存系統(tǒng)是看不見它的。必須等到訂單系統(tǒng)執(zhí)行commit請求，消息被commit后，庫存系統(tǒng)才能看到并獲取這條消息進(jìn)行后續(xù)處理。

6. half消息發(fā)送成功，但是沒收到half的響應(yīng)

以上就是RocketMQ事務(wù)消息的正向流程。

然而，還有一個問題：如果訂單系統(tǒng)發(fā)送half消息成功后卻沒有收到half消息的響應(yīng)，該如何處理呢？

在這種情況下，訂單系統(tǒng)可能會誤以為是發(fā)送half消息到MQ失敗了。訂單系統(tǒng)就會執(zhí)行回滾流程，退還支付金額，關(guān)閉訂單。

圖片

然而，此時MQ系統(tǒng)中已經(jīng)存在了一條half消息。這條half消息又該如何處理呢？

在RocketMQ中，有一套補(bǔ)償流程。RocketMQ會定期掃描處于half狀態(tài)的消息。如果一直沒有對這個消息執(zhí)行 commit/rollback 操作，超過了一定的時間，RocketMQ就會回調(diào)你的訂單系統(tǒng)的一個接口，用以確認(rèn)你本地事務(wù)的情況。

當(dāng)訂單系統(tǒng)收到MQ的回查請求時，就需要檢索一下數(shù)據(jù)庫，根據(jù)訂單狀態(tài)決定執(zhí)行commit還是rollback。

這部分邏輯就對應(yīng)上述代碼中checkLocalTransaction()方法。

圖片

7. rollback 或者 commit 失敗怎么辦？

通過上述說明，可以看到，RocketMQ是根據(jù)rollback或commit操作來決定half消息的狀態(tài)的。如果業(yè)務(wù)系統(tǒng)執(zhí)行了commit操作，則將half消息設(shè)置為可見，庫存系統(tǒng)可以消費；如果業(yè)務(wù)系統(tǒng)執(zhí)行了rollback操作，MQ就會刪除half消息。那么問題來了：如果訂單系統(tǒng)在執(zhí)行rollback或commit操作時失敗又該如何處理呢？

這時候仍然依賴于前文提到的回查機(jī)制。

由于此時MQ中的消息一直處于half狀態(tài)，超過一定的超時時間后，MQ會發(fā)現(xiàn)這個half消息有問題，然后回調(diào)你的訂單系統(tǒng)的接口。此時訂單系統(tǒng)需要根據(jù)訂單狀態(tài)來決定執(zhí)行commit請求還是rollback請求。

以上，就是RocketMQ事務(wù)消息的原理。結(jié)合文章開頭的代碼，是不是已經(jīng)很清晰了呢？

本文鏈接：http://www.www897cc.com/showinfo-26-68319-0.html實戰(zhàn)與原理：如何基于RocketMQ實現(xiàn)分布式事務(wù)？

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇： Python Pathlib模塊：一站式解決文件路徑難題

下一篇：我愛說實話，Mica-Http 超好用！

標(biāo)簽：

熱門焦點

Find N3入網(wǎng)：最高支持16+1TB

OPPO將于近期登場的Find N3折疊屏目前已經(jīng)正式入網(wǎng)，型號為PHN110。本次Find N3在外觀方面相比前兩代有很大的變化，不再是小號的橫向折疊屏，而是跟別的廠商一樣采用了較為常見的
鴻蒙OS 4.0公測機(jī)型公布：甚至連nova6都支持

華為全新的HarmonyOS 4.0操作系統(tǒng)將于今天下午正式登場，官方在發(fā)布會之前也已經(jīng)正式給出了可升級的機(jī)型產(chǎn)品，這意味著這些機(jī)型會率先支持升級享用。這次的HarmonyOS 4.0支持
印度登月最關(guān)鍵一步！月船三號今晚進(jìn)入環(huán)月軌道

8月5日消息，據(jù)印度官方消息，月船三號將于北京時間今晚21時30分左右開始近月制動進(jìn)入環(huán)月軌道。這是該探測器能夠成功的最關(guān)鍵步驟之一，如果成功將開始圍
一年經(jīng)驗在二線城市面試后端的經(jīng)驗分享

忠告這篇文章只適合2年內(nèi)工作經(jīng)驗、甚至沒有工作經(jīng)驗的朋友閱讀。如果你是2年以上工作經(jīng)驗，請果斷劃走，對你沒啥幫助~主人公這篇文章內(nèi)容來自「升職加薪」星球星友的投稿，坐
使用LLM插件從命令行訪問Llama 2

最近的一個大新聞是Meta AI推出了新的開源授權(quán)的大型語言模型Llama 2。這是一項非常重要的進(jìn)展：Llama 2可免費用于研究和商業(yè)用途。(幾小時前，swyy發(fā)現(xiàn)它已從LLaMA 2更名為Lla
使用Webdriver-manager解決瀏覽器與驅(qū)動不匹配所帶來自動化無法執(zhí)行的問題

1、前言在我們使用 Selenium 進(jìn)行 UI 自動化測試時，常常會因為瀏覽器驅(qū)動與瀏覽器版本不匹配，而導(dǎo)致自動化測試無法執(zhí)行，需要手動去下載對應(yīng)的驅(qū)動版本，并替換原有的驅(qū)動，可能還
小紅書1周漲粉49W+，我總結(jié)了小白可以用的N條漲粉筆記

作者：黃河懂運(yùn)營一條性教育視頻，被54萬人“珍藏”是什么體驗？最近，情感博主@公主是用鮮花做的，火了！僅僅憑借一條視頻，光小紅書就有超過128萬人，為她瘋狂點贊！更瘋狂的是，這
網(wǎng)傳小米汽車開始篩選交付中心建筑面積不低于3000平方米

7月7日消息，近日有微博網(wǎng)友@長三角行健者爆料稱，據(jù)經(jīng)銷商集團(tuán)反饋，小米汽車目前已經(jīng)開始了交付中心的篩選工作，要求候選場地至少有120個車位，建筑不能低
華為HarmonyOS 4升級計劃公布：首批34款機(jī)型今日開啟公測

8月4日消息，今天下午華為正式發(fā)布了HarmonyOS 4系統(tǒng)，在更流暢的前提下，還帶來了不少新功能，UI設(shè)計也有變化，會讓手機(jī)煥然一新。華為宣布，首批機(jī)型將會在

日韩成人免费在线_国产成人一二_精品国产免费人成电影在线观..._日本一区二区三区久久久久久久久不

實戰(zhàn)與原理：如何基于RocketMQ實現(xiàn)分布式事務(wù)？

使用事務(wù)消息

事務(wù)消息的原理

1. 發(fā)送half消息，探測MQ是否正常

2. half消息發(fā)送失敗

3. half消息成功，訂單系統(tǒng)執(zhí)行自己的業(yè)務(wù)邏輯

4. 訂單本地事務(wù)執(zhí)行失敗

5. 訂單本地事務(wù)執(zhí)行成功

6. half消息發(fā)送成功，但是沒收到half的響應(yīng)

7. rollback 或者 commit 失敗怎么辦？

Find N3入網(wǎng)：最高支持16+1TB

鴻蒙OS 4.0公測機(jī)型公布：甚至連nova6都支持

印度登月最關(guān)鍵一步！月船三號今晚進(jìn)入環(huán)月軌道

一年經(jīng)驗在二線城市面試后端的經(jīng)驗分享

使用LLM插件從命令行訪問Llama 2

使用Webdriver-manager解決瀏覽器與驅(qū)動不匹配所帶來自動化無法執(zhí)行的問題

小紅書1周漲粉49W+，我總結(jié)了小白可以用的N條漲粉筆記

網(wǎng)傳小米汽車開始篩選交付中心建筑面積不低于3000平方米

華為HarmonyOS 4升級計劃公布：首批34款機(jī)型今日開啟公測

最新推薦

猜你喜歡

熱門推薦

相關(guān)資訊