文章目錄? RocketMQ消息冪等閉環(huán)底層重試根源與數(shù)據(jù)庫(kù)Redis企業(yè)級(jí)落地 文章摘要 核心基礎(chǔ)底層結(jié)構(gòu)與物理模型 核心原理機(jī)制拆解與失效本質(zhì)?? 維度一生產(chǎn)者發(fā)送時(shí)的消息重復(fù)?? 維度二消費(fèi)者 ACK 丟失引發(fā)的重投?? 維度三Consumer Rebalance 期間的邊界污染 為什么常規(guī)判重會(huì)失效高并發(fā)下的漏洞 性能優(yōu)化應(yīng)用本質(zhì)與影響? 企業(yè)級(jí)“數(shù)據(jù)庫(kù) Redis”通用冪等落地策略? 核心落地策略步驟 核心代碼落地示例 冪等流水表 DDLMySQL 示例 核心字段設(shè)計(jì)與生產(chǎn)避坑解析 搭配使用的最佳實(shí)踐建議? 面試回答思路結(jié)構(gòu)化高分話術(shù)? RocketMQ消息冪等閉環(huán)底層重試根源與數(shù)據(jù)庫(kù)Redis企業(yè)級(jí)落地 文章摘要在分布式系統(tǒng)中消息隊(duì)列為保證可靠性普遍采用“至少一次At-Least-Once”投遞語(yǔ)義這直接導(dǎo)致消息重復(fù)消費(fèi)成為必然。RocketMQ 從底層架構(gòu)上無(wú)法全局消滅重復(fù)其根源在于網(wǎng)絡(luò)閃斷重試、ACK 丟失補(bǔ)償及 Consumer Rebalance 機(jī)制。實(shí)現(xiàn)消費(fèi)冪等的底層核心在于將消息消費(fèi)轉(zhuǎn)化為具備“冪等性”的狀態(tài)機(jī)躍遷通常依托業(yè)務(wù)唯一 Key、Redis 分布式鎖與數(shù)據(jù)庫(kù)唯一索引在緩存與存儲(chǔ)層筑牢防線實(shí)現(xiàn)高并發(fā)下的數(shù)據(jù)絕對(duì)一致。 核心基礎(chǔ)底層結(jié)構(gòu)與物理模型在分布式消息模型中為了防止數(shù)據(jù)丟失系統(tǒng)采用的是At-Least-Once至少一次投遞策略。這意味著消息可能會(huì)被重復(fù)發(fā)送給消費(fèi)者。從 RocketMQ 的底層存儲(chǔ)與消費(fèi)模型來(lái)看重復(fù)消息的物理根源交織在以下幾個(gè)核心組件中ConsumeQueue 索引模型Consumer 通過(guò)維護(hù)自身的Offset消費(fèi)進(jìn)度去拉取 CommitLog 中的消息。Offset 的提交與業(yè)務(wù)消費(fèi)成功之間并非絕對(duì)的原子操作。ConsumeQueue就像外賣取餐柜?你消費(fèi)者是按柜子上的取餐號(hào)Offset來(lái)拿外賣消息的。規(guī)則是你得先把外賣送到顧客手里業(yè)務(wù)消費(fèi)成功再去系統(tǒng)里標(biāo)記「這個(gè)號(hào)的餐我送完了」提交Offset。但這兩步不是綁死的——萬(wàn)一你剛把餐送到還沒來(lái)得及點(diǎn)確認(rèn)系統(tǒng)就以為你這單沒送轉(zhuǎn)頭又把同一餐派給你了。客戶端 Offset 異步同步Consumer 消費(fèi)完消息后通常采用定時(shí)或異步的方式向 Broker 提交 Offset。如果在此期間客戶端發(fā)生 Crash、OOM 或網(wǎng)絡(luò)抖動(dòng)未及時(shí)同步的 Offset 會(huì)導(dǎo)致重啟后重新拉取相同區(qū)間的消息。?異步提交Offset就像下班前統(tǒng)一簽考勤?你不是送完一單就立刻在系統(tǒng)里點(diǎn)確認(rèn)而是攢一批、定時(shí)統(tǒng)一上報(bào)簽到。要是剛送完幾單還沒來(lái)得及統(tǒng)一打卡你手機(jī)突然沒電關(guān)機(jī)了客戶端Crash/OOM或者路上信號(hào)斷了網(wǎng)絡(luò)抖動(dòng)這些沒來(lái)得及簽到的單子等你第二天一上線系統(tǒng)又會(huì)原封不動(dòng)再給你派一遍。Rebalance負(fù)載均衡機(jī)制當(dāng)消費(fèi)者實(shí)例發(fā)生上下線變化時(shí)Queue 會(huì)在不同 Consumer 之間重新分配。由于舊實(shí)例的消費(fèi)進(jìn)度未完全持久化或新實(shí)例拉取起點(diǎn)存在偏差極端情況下會(huì)導(dǎo)致邊界消息被重復(fù)讀取。Rebalance就像站點(diǎn)臨時(shí)調(diào)班分單?站點(diǎn)消費(fèi)組突然有人請(qǐng)假、有人新入職消費(fèi)者上下線站長(zhǎng)就得把手里的外賣片區(qū)重新分給大家。萬(wàn)一之前負(fù)責(zé)這片的小哥沒來(lái)得及把最后幾單的送達(dá)記錄同步給站長(zhǎng)新接手的小哥從站點(diǎn)記錄的進(jìn)度開始派單就會(huì)把上一個(gè)人已經(jīng)送過(guò)的那幾單又給顧客再送一遍。Broker (CommitLog / ConsumeQueue) │ ├──(1. 投遞消息)── Consumer 1 (處理業(yè)務(wù)如扣減庫(kù)存) │ │ │ └──(2. 異常閃斷 / 異步 Offset 提交延遲) │ └──(3. 重平衡 / 重試)── Consumer 2 (再次拉取到相同 Offset 消息 ? 發(fā)生重復(fù)消費(fèi)) 核心原理機(jī)制拆解與失效本質(zhì)從“引擎視角”來(lái)看為什么 RocketMQ 無(wú)法在 Broker 層自動(dòng)實(shí)現(xiàn)全局冪等核心原因在于狀態(tài)爆炸與成本權(quán)衡。如果 Broker 要攔截所有重復(fù)消息必須在內(nèi)存或磁盤中維護(hù)全量的歷史索引這會(huì)帶來(lái)災(zāi)難性的內(nèi)存開銷和存儲(chǔ)放大。因此冪等的防線必須下沉至消費(fèi)端其底層觸發(fā)場(chǎng)景可拆解為三個(gè)典型維度?? 維度一生產(chǎn)者發(fā)送時(shí)的消息重復(fù)當(dāng)一條消息已被成功發(fā)送到 RocketMQ 的 Broker 中并完成磁盤持久化此時(shí)出現(xiàn)了網(wǎng)絡(luò)閃斷或者生產(chǎn)者宕機(jī)導(dǎo)致 Broker 對(duì)生產(chǎn)者應(yīng)答失敗。生產(chǎn)者若意識(shí)到消息發(fā)送失敗并嘗試再次發(fā)送消費(fèi)者后續(xù)會(huì)收到兩條內(nèi)容相同且Message ID相同的消息導(dǎo)致 Consumer 被動(dòng)消費(fèi)兩次。?? 維度二消費(fèi)者 ACK 丟失引發(fā)的重投消息已投遞到 Consumer 并完成業(yè)務(wù)處理但在向 Broker 返回消費(fèi)成功 ACK 確認(rèn)響應(yīng)時(shí)發(fā)生網(wǎng)絡(luò)閃斷導(dǎo)致 Broker 未能成功收到響應(yīng)。Broker 認(rèn)為 Consumer 未能消費(fèi)成功為了保證消息至少被消費(fèi)一次將在網(wǎng)絡(luò)恢復(fù)后再次嘗試投遞之前已被處理過(guò)的消息。?? 維度三Consumer Rebalance 期間的邊界污染當(dāng) Broker 重啟或 Consumer 擴(kuò)容、縮容觸發(fā)重新負(fù)載均衡時(shí)Consumer 讀取 Broker 中的 offset 可能還沒及時(shí)更新從而收到曾經(jīng)被消費(fèi)過(guò)的消息。 為什么常規(guī)判重會(huì)失效高并發(fā)下的漏洞Message ID 沖突隱患RocketMQ 的Message ID在特定集群環(huán)境下可能出現(xiàn)沖突因此真正安全的冪等處理絕不能以 Message ID 作為處理依據(jù)而必須依靠業(yè)務(wù)層生成的全局唯一標(biāo)識(shí)Message Key。先查詢后插入的并發(fā)穿透如果開發(fā)者在消費(fèi)時(shí)簡(jiǎn)單采用“先 SELECT 檢查是否存在再 INSERT”的邏輯在多線程或高并發(fā)集群下兩條相同的消息可能同時(shí)穿透查詢導(dǎo)致并發(fā)沖突或臟數(shù)據(jù)。 性能優(yōu)化應(yīng)用本質(zhì)與影響從架構(gòu)演進(jìn)的視角來(lái)看冪等設(shè)計(jì)本質(zhì)上是用存儲(chǔ)鎖競(jìng)爭(zhēng)與額外的網(wǎng)絡(luò)/計(jì)算開銷來(lái)?yè)Q取分布式系統(tǒng)的數(shù)據(jù)絕對(duì)一致性。? 企業(yè)級(jí)“數(shù)據(jù)庫(kù) Redis”通用冪等落地策略從架構(gòu)演進(jìn)的視角來(lái)看,冪等設(shè)計(jì)本質(zhì)上是用存儲(chǔ)鎖競(jìng)爭(zhēng)與額外的網(wǎng)絡(luò)/計(jì)算開銷來(lái)?yè)Q取分布式系統(tǒng)的數(shù)據(jù)絕對(duì)一致性。為了兼顧高性能與絕對(duì)準(zhǔn)確性,生產(chǎn)環(huán)境通常采用多級(jí)校驗(yàn)Redis 緩存防線 數(shù)據(jù)庫(kù)唯一鍵兜底的通用解決方案。針對(duì)第三層原子狀態(tài)落地與事務(wù)保證,如果盲目追求在同一個(gè)Transactional事務(wù)里同時(shí)操作 Redis 和 DB由于 Redis 不支持 XA 協(xié)議會(huì)導(dǎo)致緩存臟數(shù)據(jù)或不一致是不可行的。因此業(yè)界標(biāo)準(zhǔn)的工程實(shí)現(xiàn)采用的是“先提交 DB后更新 Redis” 數(shù)據(jù)庫(kù)唯一索引約束。? 核心落地策略步驟第一層Redis 快速攔截Consumer 消費(fèi)消息時(shí)拿到唯一的業(yè)務(wù)標(biāo)識(shí)消息 Key,首先去 Redis 緩存中查詢是否存在對(duì)應(yīng)的記錄。如果存在,說(shuō)明本次操作是重復(fù)性操作,直接攔截。第二層數(shù)據(jù)庫(kù)防穿透校驗(yàn)與唯一索引兜底利用數(shù)據(jù)庫(kù)表的唯一約束Unique Key處理并發(fā)穿透。多個(gè)線程同時(shí)寫入時(shí)數(shù)據(jù)庫(kù)引擎的底層鎖會(huì)強(qiáng)制攔截只允許一個(gè)成功其余拋出DuplicateKeyException。第三層原子狀態(tài)落地與事務(wù)保證將核心業(yè)務(wù)如扣減庫(kù)存與冪等流水表插入放在同一個(gè)本地事務(wù)中。采用先提交 DB后更新 Redis的策略DB 事務(wù)提交成功后再更新 Redis 緩存。即使 Redis 更新失敗下次重試依然能從 DB 兜底絕不會(huì)發(fā)生數(shù)據(jù)不一致。 核心代碼落地示例mqConsumer.registerMessageListener(newMessageListenerConcurrently(){OverridepublicConsumeConcurrentlyStatusconsumeMessage(ListMessageExtmsgs,ConsumeConcurrentlyContextcontext){for(MessageExtmsg:msgs){// 獲取業(yè)務(wù)唯一標(biāo)識(shí) KeyStringkeymsg.getKeys();try{// 1. Redis 快速攔截前置性能過(guò)濾Objectobjredis.get(key);if(null!obj){logger.info(Redis 攔截消息重復(fù)消費(fèi)Key: {},key);continue;}// 2. 數(shù)據(jù)庫(kù)事務(wù)執(zhí)行包含唯一鍵冪等校驗(yàn)與業(yè)務(wù)處理messageService.handleBusinessAndSaveLog(msg,key);// 3. 【關(guān)鍵】DB 事務(wù)成功提交后才去回填 Redis 緩存redis.set(key,SUCCESS,Duration.ofHours(24));}catch(DuplicateKeyExceptione){// 4. 捕獲數(shù)據(jù)庫(kù)唯一鍵沖突異常視為重復(fù)消息處理靜默返回成功logger.warn(唯一索引沖突DB兜底消息重復(fù)消費(fèi), Key: {},key);}catch(Exceptione){logger.error(消息消費(fèi)異常等待重試, Key: {},key,e);returnConsumeConcurrentlyStatus.RECONSUME_LATER;}}returnConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});其中messageService.handleBusinessAndSaveLog(msg, key)的內(nèi)部事務(wù)實(shí)現(xiàn)Transactional(rollbackForException.class)publicvoidhandleBusinessAndSaveLog(MessageExtmsg,Stringkey){// a. 嘗試插入冪等流水表該 key 字段必須設(shè)置 UNIQUE INDEX 唯一索引// 如果重復(fù)投遞這里直接拋出 DuplicateKeyException 并觸發(fā)事務(wù)回滾messageIdempotencyDao.insert(key,PROCESSING,LocalDateTime.now());// b. 核心業(yè)務(wù)處理例如扣減庫(kù)存stockService.decrease(msg.getPayload());// c. 更新流水狀態(tài)為成功messageIdempotencyDao.updateStatus(key,SUCCESS);}在企業(yè)級(jí)落地“Redis 數(shù)據(jù)庫(kù)唯一索引”的冪等方案時(shí)冪等流水表通常也叫消息消費(fèi)流水表 / 冪等防重表是支撐數(shù)據(jù)庫(kù)兜底防線的核心載體。以下是生產(chǎn)環(huán)境中最推薦的表結(jié)構(gòu)設(shè)計(jì)及建表 SQL同時(shí)附帶了關(guān)鍵字段的設(shè)計(jì)意圖解析 冪等流水表 DDLMySQL 示例CREATETABLEmessage_idempotency_log(idBIGINTNOTNULLAUTO_INCREMENTCOMMENT自增主鍵,msg_keyVARCHAR(128)NOTNULLCOMMENT業(yè)務(wù)唯一消息 Key核心防重字段必須建立唯一索引,topicVARCHAR(64)NOTNULLCOMMENTRocketMQ Topic 名稱便于多業(yè)務(wù)線復(fù)用或隔離,statusVARCHAR(32)NOTNULLCOMMENT消費(fèi)狀態(tài)PROCESSING(處理中), SUCCESS(成功), FAIL(失敗),remarkVARCHAR(255)DEFAULTNULLCOMMENT備注或錯(cuò)誤信息消費(fèi)失敗時(shí)記錄異常原因,create_timeDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMPCOMMENT創(chuàng)建時(shí)間,update_timeDATETIMENOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMPCOMMENT更新時(shí)間,PRIMARYKEY(id),UNIQUEKEYuk_msg_key(msg_key)USINGBTREECOMMENT業(yè)務(wù) Key 唯一索引實(shí)現(xiàn)高并發(fā)下數(shù)據(jù)庫(kù)兜底防線的核心)ENGINEInnoDBDEFAULTCHARSETutf8mb4COMMENTMQ 消息消費(fèi)冪等流水防重表; 核心字段設(shè)計(jì)與生產(chǎn)避坑解析msg_key業(yè)務(wù)唯一標(biāo)識(shí)作用這是整張表的靈魂。它必須傳入業(yè)務(wù)層生成的全局唯一 Key例如訂單號(hào) 業(yè)務(wù)類型或業(yè)務(wù)流水號(hào)而絕對(duì)不能直接拿 RocketMQ 自帶的Message ID。唯一索引UNIQUE KEY uk_msg_key這是整個(gè)架構(gòu)的“終極保鏢”。當(dāng)高并發(fā)或緩存穿透導(dǎo)致兩條相同消息同時(shí)嘗試寫入時(shí)數(shù)據(jù)庫(kù)引擎會(huì)通過(guò)這個(gè)唯一索引強(qiáng)制攔截其中一個(gè)并拋出DuplicateKeyException。status消費(fèi)狀態(tài)機(jī)PROCESSING處理中當(dāng)消息剛進(jìn)入事務(wù)準(zhǔn)備執(zhí)行業(yè)務(wù)時(shí)寫入的狀態(tài)。SUCCESS成功業(yè)務(wù)執(zhí)行完畢后更新的狀態(tài)。為什么要有PROCESSING在極少數(shù)極端情況下如服務(wù)在業(yè)務(wù)執(zhí)行中途突然 OOM 崩潰流水表里會(huì)留下一個(gè)PROCESSING狀態(tài)的臟數(shù)據(jù)。通過(guò)配合定時(shí)任務(wù)或超時(shí)檢查機(jī)制系統(tǒng)可以識(shí)別出哪些消息“卡死”了從而進(jìn)行人工介入或補(bǔ)償重試。topic主題隔離可選擴(kuò)展如果你們系統(tǒng)里有多個(gè)不同的 Topic 共用一張流水表加上topic字段可以防止不同業(yè)務(wù)線偶然生成的msg_key發(fā)生碰撞。如果是一個(gè)系統(tǒng)一張表也可以直接省略。 搭配使用的最佳實(shí)踐建議索引優(yōu)化因?yàn)閙sg_key已經(jīng)加了UNIQUE INDEX數(shù)據(jù)庫(kù)會(huì)自動(dòng)為其建立 B 樹索引因此根據(jù) Key 的查詢性能極高毫秒級(jí)完全不用擔(dān)心引入流水表會(huì)導(dǎo)致查詢變慢。數(shù)據(jù)清理策略歸檔/分表隨著業(yè)務(wù)量增長(zhǎng)冪等流水表的數(shù)據(jù)量會(huì)迅速膨脹。生產(chǎn)環(huán)境中通常會(huì)設(shè)置數(shù)據(jù)保留期例如保留 7 天或 15 天通過(guò)定時(shí)任務(wù)清理過(guò)期的SUCCESS狀態(tài)流水或者按月進(jìn)行分庫(kù)分表。? 面試回答思路結(jié)構(gòu)化高分話術(shù)面試官“RocketMQ 保證的是至少一次投遞下游消費(fèi)時(shí)怎么保證冪等性你們?cè)谏a(chǎn)中是怎么落地的如何處理原子性”三步走高分回答定基調(diào)“面試官分布式消息隊(duì)列基于網(wǎng)絡(luò)不可靠性采用的是‘At-Least-Once至少一次’投遞語(yǔ)義重復(fù)消費(fèi)是必然發(fā)生的。RocketMQ 從架構(gòu)設(shè)計(jì)上無(wú)法在 Broker 層做全局冪等因?yàn)闋顟B(tài)維護(hù)成本太高因此冪等設(shè)計(jì)是消費(fèi)端的必修課?！敝v本質(zhì)“從底層引擎與投遞場(chǎng)景來(lái)看重復(fù)消費(fèi)主要由生產(chǎn)者重試、ACK 丟失重投以及Consumer Rebalance 引起的進(jìn)度重置導(dǎo)致。由于Message ID存在沖突風(fēng)險(xiǎn)我們必須強(qiáng)制綁定業(yè)務(wù)的唯一Message Key作為冪等憑證?！闭劶夹g(shù)方案與落地突出原子性閉環(huán)“在生產(chǎn)環(huán)境中我們采用的是‘Redis 緩存前置攔截 數(shù)據(jù)庫(kù)唯一索引兜底’的組合拳首先通過(guò) Redis 高性能攔截絕大部分重復(fù)流量針對(duì)緩存過(guò)期或穿透依托數(shù)據(jù)庫(kù)表的唯一約束Unique Key進(jìn)行強(qiáng)校驗(yàn)。在本地Transactional事務(wù)中我們將冪等流水表插入與核心業(yè)務(wù)綁定若觸發(fā)DuplicateKeyException證明歷史已處理過(guò)直接靜默返回CONSUME_SUCCESS在狀態(tài)落地時(shí)我們堅(jiān)持‘先提交 DB后更新 Redis’的原則。把數(shù)據(jù)庫(kù)作為保障數(shù)據(jù)一致性的最終真理Redis 僅作為加速緩存。這樣既避免了分布式事務(wù)的復(fù)雜性又完美兼顧了系統(tǒng)高吞吐量與數(shù)據(jù)強(qiáng)一致性。”