title: 把 Kafka 當(dāng)隊(duì)列用丟了 0.3% 的消息Kafka 與 RocketMQ 在可靠性、順序、事務(wù)上的 4 筆真實(shí)賬date: 2026-08-22category: 消息隊(duì)列tags: [Java, 后端, 消息隊(duì)列, Kafka, RocketMQ, 微服務(wù)]我們交易系統(tǒng)的異步鏈路一開(kāi)始全用 Kafka。理由很樸素社區(qū)火、吞吐高、大家都用。直到一次對(duì)賬發(fā)現(xiàn)有 0.3% 的支付成功消息下游沒(méi)收到導(dǎo)致訂單狀態(tài)卡在「支付中」。排查下來(lái)不是 Kafka 的 bug而是我們用錯(cuò)了它的可靠性模型——Kafka 默認(rèn)是「不丟 broker 端」但「不保證消費(fèi)端一定處理成功」。這篇不聊誰(shuí)更牛只聊我們實(shí)打?qū)嵲谶@倆隊(duì)列上花的 4 筆賬可靠性、順序、事務(wù)、堆積。選型翻車往往是因?yàn)槟阈枰哪且豁?xiàng)正好是它弱的那一項(xiàng)。第一筆賬可靠性Kafka 默認(rèn) at-least-onceKafka 的生產(chǎn)者默認(rèn)acks1leader 收到就返回broker 宕機(jī)可能丟消息。要「不丟」得acksall 副本min.insync.replicas1// Kafka 生產(chǎn)者要 broker 端不丟得這么配 Properties p new Properties(); p.put(acks, all); // 等所有 ISR 副本確認(rèn) p.put(min.insync.replicas, 2); // 至少 2 個(gè)副本落盤(pán) p.put(enable.idempotence, true); // 開(kāi)啟冪等避免重試導(dǎo)致重復(fù) p.put(retries, 5); ProducerString, String producer new KafkaProducer(p);但即便 broker 不丟消費(fèi)端也可能丟Kafka 是「拉取 手動(dòng)提交 offset」模型如果你處理完業(yè)務(wù)邏輯、還沒(méi)提交 offset 就崩了這條消息會(huì)被重新投遞——這是 at-least-once。我們那次丟消息的根因是反過(guò)來(lái)的為了快我們先提交 offset 再處理結(jié)果處理邏輯拋異常消息算「已消費(fèi)」永遠(yuǎn)丟了。// 錯(cuò)誤示范先 commit 再處理處理失敗消息就沒(méi)了 consumer.commitSync(); // offset 已提交 process(record); // 這里拋異常消息丟失 // 正確做法先處理再提交用冪等兜底重復(fù) process(record); consumer.commitSync(); // 處理成功才提交RocketMQ 默認(rèn)是at-least-once消費(fèi)重試隊(duì)列消費(fèi)失敗會(huì)進(jìn)重試隊(duì)列默認(rèn) 16 次還不行進(jìn)死信隊(duì)列消息不會(huì)憑空消失。這對(duì)「消息不能丟」的業(yè)務(wù)更友好。第二筆賬順序消息Kafka 靠分區(qū)、RocketMQ 有原生支持我們需要「同一個(gè)訂單的支付、發(fā)貨、完成消息嚴(yán)格有序」。Kafka 只能保證單個(gè)分區(qū)內(nèi)有序所以得把同一個(gè)訂單 ID 哈希到同一個(gè) partition// Kafka相同 orderId 進(jìn)同一 partition靠分區(qū)器保證局部順序 producer.send(new ProducerRecord( order-topic, orderId.hashCode() % partitionCount, // key 決定分區(qū) orderId, payload));但代價(jià)是一旦某個(gè)分區(qū)消費(fèi)慢會(huì)阻塞整個(gè)分區(qū)的后續(xù)消息Kafka 單分區(qū)是串行消費(fèi)的。RocketMQ 則提供「順序消息」語(yǔ)義broker 端用分段鎖保證同一 queue 的消息順序消費(fèi)者也順序拉取語(yǔ)義更省心。我們訂單狀態(tài)機(jī)用 RocketMQ 順序消息后省掉了自己維護(hù)「分區(qū)擁堵監(jiān)控」的那套告警。第三筆賬事務(wù)消息這是兩者分水嶺我們要「本地訂單落庫(kù)」和「發(fā)支付消息」要么都成、要么都敗。Kafka 有事務(wù) API但用起來(lái)很重// Kafka 事務(wù)消息要 encloser id initTransactions producer.initTransactions(); try { producer.beginTransaction(); producer.send(new ProducerRecord(order-topic, payload)); // 本地事務(wù)這里要自己保證Kafka 事務(wù)只管消息 orderDb.insert(order); producer.commitTransaction(); } catch (Exception e) { producer.abortTransaction(); }Kafka 的事務(wù)本質(zhì)是「把多條消息原子提交」它不保證和你的本地 DB 事務(wù)綁定——中間還是可能「DB 成了、消息回滾」或反過(guò)來(lái)要用最大努力通知或額外對(duì)賬兜底。RocketMQ 的事務(wù)消息則專門(mén)為此設(shè)計(jì)半消息 回查機(jī)制broker 會(huì)主動(dòng)問(wèn)你「本地事務(wù)到底成了沒(méi)」// RocketMQ 事務(wù)消息實(shí)現(xiàn) TransactionListener 回查本地事務(wù)狀態(tài) TransactionMQProducer producer new TransactionMQProducer(order_group); producer.setTransactionListener(new TransactionListener() { Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { // 執(zhí)行本地 DB 事務(wù) return orderDb.insert(order) ? COMMIT_MESSAGE : ROLLBACK_MESSAGE; } Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { // broker 回查去 DB 查這筆訂單到底在不在 return orderDb.exists(msg.getKeys()) ? COMMIT_MESSAGE : UNKNOW; } }); producer.sendMessageInTransaction(new Message(order-topic, payload.getBytes()), null);這筆賬我們交了學(xué)費(fèi)用 Kafka 硬做事務(wù)最后還是靠每小時(shí)對(duì)賬補(bǔ)單多寫(xiě)了 200 行對(duì)賬代碼。后來(lái)涉及「DB 消息」強(qiáng)一致的鏈路遷到 RocketMQ 事務(wù)消息對(duì)賬代碼刪了一大半。第四筆賬延遲消息RocketMQ 開(kāi)箱即用還有一個(gè)我們真實(shí)用到的場(chǎng)景訂單創(chuàng)建后 30 分鐘未支付自動(dòng)關(guān)單。RocketMQ 原生支持延遲消息// RocketMQ指定延遲級(jí)別18 級(jí)覆蓋 1s~2h Message msg new Message(order-close-topic, payload.getBytes()); msg.setDelayTimeLevel(16); // 16 級(jí) 30 分鐘 producer.send(msg);Kafka 沒(méi)有原生延遲消息我們得自己搞「時(shí)間輪 外部存儲(chǔ)」或者「建多個(gè)延遲 topic 定時(shí)搬運(yùn)」多了一套組件要維護(hù)。如果你業(yè)務(wù)里關(guān)單、重試、定時(shí)通知這類場(chǎng)景多RocketMQ 的延遲消息能省不少事。第五筆賬堆積與吞吐Kafka 仍是天花板反過(guò)來(lái)純吞吐和堆積量Kafka 確實(shí)更強(qiáng)。我們埋點(diǎn)日志單條小、允許少量重復(fù)、不需要順序用 Kafka單集群輕松扛百萬(wàn) TPS堆積上億條也沒(méi)壓力。RocketMQ 在「功能豐富」延遲消息、事務(wù)、順序、重試上更全但極端吞吐略遜。維度KafkaRocketMQ默認(rèn)可靠性at-least-once需配 acksallat-least-once 重試/死信順序消息單分區(qū)有序靠 key 哈希原生順序消息語(yǔ)義事務(wù)消息API 重不綁 DB 事務(wù)半消息 回查綁 DB 友好延遲消息無(wú)原生需自建18 級(jí)原生支持堆積/吞吐極高高功能更全消費(fèi)端冪等兩個(gè)隊(duì)列都需要的兜底不管選 Kafka 還是 RocketMQat-least-once 語(yǔ)義都意味著「消息可能重復(fù)投送」。所以消費(fèi)端必須自己做冪等這是繞不開(kāi)的。我們用「消息唯一鍵 Redis 去重」兜底// 消費(fèi)冪等同一 messageId 只處理一次 public void consume(String messageId, String payload) { // SETNXkey 不存在才設(shè)成功返回 1 表示首次處理 Long r redis.opsForValue() .setIfAbsent(msg: messageId, 1, 24, TimeUnit.HOURS); if (r null || r 0) { return; // 重復(fù)消息直接丟棄 } businessProcess(payload); // 真正處理 }這個(gè)冪等層對(duì)兩個(gè)隊(duì)列都通用也正是前面說(shuō)的「用 Kafka 做關(guān)鍵隊(duì)列要自己補(bǔ)的兜底」之一。RocketMQ 雖然自帶重試/死信但重試過(guò)來(lái)的消息 messageId 會(huì)變?nèi)ブ氐糜脴I(yè)務(wù)自己的唯一鍵比如訂單 ID不能依賴消息系統(tǒng)給的 ID。這點(diǎn)踩過(guò)我們一度用 RocketMQ 的 msgId 去重結(jié)果重試消息 msgId 不同去重失效還是重復(fù)處理了。我的選型判斷別再問(wèn)「Kafka 和 RocketMQ 誰(shuí)更好」——這問(wèn)題本身錯(cuò)。我們現(xiàn)在的架構(gòu)是混用日志、埋點(diǎn)、流式計(jì)算 → Kafka要的是吞吐和生態(tài)Flink 集成好。交易、訂單、履約這類「消息不能丟、要順序、要事務(wù)」的核心鏈路 → RocketMQ它把可靠性相關(guān)的坑都幫你填了。踩完 0.3% 丟消息這個(gè)坑后我形成了一個(gè)觀點(diǎn)用 Kafka 做關(guān)鍵業(yè)務(wù)隊(duì)列你得自己補(bǔ)冪等、補(bǔ)對(duì)賬、補(bǔ)重試等于自己造半個(gè) RocketMQ。如果團(tuán)隊(duì)人力緊張、業(yè)務(wù)又關(guān)鍵直接用 RocketMQ 省下的排障時(shí)間遠(yuǎn)比那點(diǎn)吞吐差距值錢(qián)。只有當(dāng)你真的需要 Kafka 的生態(tài)流處理、海量日志時(shí)才值得為它額外寫(xiě)那些兜底代碼。復(fù)盤(pán)數(shù)字那次丟消息持續(xù)約 9 小時(shí)夜間批處理 白天高峰疊加最終對(duì)賬補(bǔ)單 1.2 萬(wàn)筆客訴 37 起。把交易鏈路遷到 RocketMQ 事務(wù)消息 消費(fèi)冪等后3 個(gè)月同類「消息丟失」告警為零而日志鏈路繼續(xù)用 Kafka單日吞吐穩(wěn)定在 800 萬(wàn)條以上堆積峰值 2 億條時(shí)消費(fèi)無(wú)阻塞?；煊弥蠛诵逆溌费a(bǔ)單率從 0.3% 降到萬(wàn)分之一以下剩余的是業(yè)務(wù)側(cè)自身重試。思考題RocketMQ 的事務(wù)消息用「半消息 回查」解決 DB 與消息的一致性但回查本身有次數(shù)上限如果本地事務(wù)卡死、回查一直返回 UNKNOW這條半消息會(huì)怎樣你會(huì)在業(yè)務(wù)側(cè)怎么兜底