不想引戰(zhàn)但我要先把話說清楚RabbitMQ 的 Event Exchange Plugin文件名叫rabbitmq-event-exchange絕對是我用過的最“冷門但真香”的官方插件之一。你一搜 RabbitMQ 教程滿屏都是普通 exchange、死信隊(duì)列、延遲隊(duì)列但很少有人告訴你RabbitMQ 還能直接把消息落到 MongoDB 里然后用 HTTP 接口讓消費(fèi)者像“翻聊天記錄”一樣去拉消息。這篇文章我就拿自己在測試環(huán)境里的完整折騰記錄說話把 Event Exchange Plugin 的設(shè)計(jì)思路、核心原理、部署步驟、消費(fèi)代碼和一堆坑一次性講透。文章里需要動手的地方我都給了可以直接套用的配置和 Demo適合對 RabbitMQ 已經(jīng)有一定基礎(chǔ)、又想擴(kuò)展消息隊(duì)列形態(tài)的開發(fā)者閱讀。如果你完全沒用過 RabbitMQ建議先補(bǔ)一補(bǔ) exchange、queue、binding 的基本概念再回來看這篇效果會好很多。1. 這個東西到底是干嘛的先說需求再談方案1.1 什么樣的場景會讓人想去翻這個插件我先講一個自己遇到過的真實(shí)需求。某段時(shí)間我在做一個內(nèi)部審計(jì)系統(tǒng)需要把所有用戶操作事件登錄、下單、改配置、導(dǎo)數(shù)據(jù)全部記錄下來而且要求這些事件“不能丟、不能改、還能從頭重放”。如果用 RabbitMQ 普通隊(duì)列消息被消費(fèi)者拉走之后就從 broker 里刪了想回放沒門。如果直接把事件寫 MySQL寫入壓力一大就卡業(yè)務(wù)而且消費(fèi)端想要“按游標(biāo)增量拉取”還得自己設(shè)計(jì) offset 管理。折騰了一圈直到我翻 RabbitMQ 官方插件列表時(shí)看到rabbitmq-event-exchange才真正找到順手的方案。這個插件的定位很明確它為 RabbitMQ 增加了一種新的交換器類型x-event。當(dāng)你把消息發(fā)到x-event類型的交換器上時(shí)消息并不會像普通交換器那樣被路由到某個 AMQP 隊(duì)列而是被持久化寫入 MongoDB 的指定集合里。消費(fèi)者不需要聲明 AMQP 隊(duì)列也不需要走 AMQP 協(xié)議去消費(fèi)而是通過 HTTP 接口從插件暴露的端點(diǎn)拉取這一類“事件流”。簡單說這個插件把 RabbitMQ 從一個“消息分發(fā)中心”變成了一個“事件流存儲 分發(fā)中心”。對比普通隊(duì)列它最打動我的三個點(diǎn)一是消息默認(rèn)全量落庫天然支持重放二是消費(fèi)者之間通過 MongoDB 游標(biāo)隔離進(jìn)度不會互相踢掉未消費(fèi)的數(shù)據(jù)三是消費(fèi)者通過 HTTP 拉取跨語言、跨防火墻都很好接。1.2 這個插件到底解決了什么問題我把問題收斂成三點(diǎn)你對照自己的項(xiàng)目看有沒有共鳴事件需要可重放普通 AMQP 隊(duì)列對已消費(fèi)消息不保留想重新跑一遍歷史事件做數(shù)據(jù)訂正非常費(fèi)勁。多消費(fèi)者各自推進(jìn)你希望每個消費(fèi)者都拿到全量事件而不是一堆消費(fèi)者搶一條消息Work Queue 模式。事件流場景需要的其實(shí)是 Pub/Sub 獨(dú)立游標(biāo)。不想引入額外的重系統(tǒng)很多團(tuán)隊(duì)一提到事件流就想到 Kafka但 Kafka 部署和運(yùn)維成本對小團(tuán)隊(duì)來說并不輕。如果基礎(chǔ)設(shè)施里本來就有 MongoDB那用這個插件就能低成本獲得類似“按偏移量消費(fèi)”的能力。所以這個插件適合誰我認(rèn)為適合這幾類人正在做事件溯源Event Sourcing架構(gòu)的開發(fā)者需要給系統(tǒng)搭建操作審計(jì)日志的團(tuán)隊(duì)想把 RabbitMQ 消息轉(zhuǎn)到 MongoDB 做長期歸檔和分析的中間件維護(hù)者以及想學(xué)習(xí) RabbitMQ 擴(kuò)展機(jī)制、想研究“交換器類型擴(kuò)展點(diǎn)”源碼的人。注意這個插件不等于 Kafka 的“完整平替”。它是把 MongoDB 當(dāng)作事件存儲介質(zhì)通過 HTTP 游標(biāo)方式消費(fèi)吞吐上限受 MongoDB 性能和 HTTP 拉取頻率制約。選型前一定要對大流量場景做壓測。2. 工作原理解析一條消息從發(fā)出來到被消費(fèi)都經(jīng)歷了什么2.1 事件交換器類型與綁定關(guān)系先糾正一個常見的誤解x-event并不是 RabbitMQ 自帶的交換器類型而是這個插件通過 Erlang 代碼注冊進(jìn)去的自定義交換器類型。插件啟用后你可以在管理后臺的 Exchange 列表里看到類型為x-event的交換器。這個交換器的行為很特殊。普通direct、topic、fanout交換器收到消息后會根據(jù) routing key 和 binding 把消息分發(fā)到綁定隊(duì)列然后消費(fèi)者用 AMQP 協(xié)議從隊(duì)列里取消息。而x-event交換器收到消息后會直接把消息塞進(jìn) MongoDB 的某個集合里并不關(guān)心你有沒有 AMQP 隊(duì)列。那綁定關(guān)系怎么建立這個插件使用 Mongo 數(shù)據(jù)庫里的配置集合保存“虛擬隊(duì)列”的概念。每一個虛擬隊(duì)列可以理解為 MongoDB 集合上的一個游標(biāo)位置。同一個事件流集合可以同時(shí)存在多個虛擬隊(duì)列每個虛擬隊(duì)列各自獨(dú)立推進(jìn)游標(biāo)互不影響。實(shí)際操作中建立綁定關(guān)系的姿勢和普通 RabbitMQ 不太一樣——你不太需要去聲明一個普通 AMQP 隊(duì)列然后 bind 到 exchange 上當(dāng)然插件也兼容了部分 AMQP 操作更多時(shí)候是通過插件自帶的 HTTP API 來注冊虛擬隊(duì)列綁定。這一點(diǎn)在后面第 3 章展開。2.2 Mongo 落庫的存儲格式與游標(biāo)推進(jìn)邏輯這也是我建議大家重點(diǎn)理解的地方。rabbitmq-event-exchange插件的核心設(shè)計(jì)思想是充分利用 MongoDB 自身的oplog和tailable cursor機(jī)制來實(shí)現(xiàn)事件流。生產(chǎn)者發(fā)消息到x-event交換器插件將消息體序列化后插入到 MongoDB 的某個集合collection中。每個插入的文檔會自動獲得_idObjectId 類型。MongoDB 的 ObjectId 自帶時(shí)間戳且單調(diào)遞增同一秒內(nèi)順序也遞增天然適合作為事件游標(biāo)。消費(fèi)者通過插件 HTTP API 拉取事件時(shí)插件會在 MongoDB 上開啟一個 tailable cursor拉取_id大于當(dāng)前游標(biāo)位置的新文檔然后包裝成類似 AMQP 消息的格式返回給客戶端?？蛻舳嗣看卫《紩谡埱髤?shù)里帶上自己處理到的游標(biāo)值eventId或類似字段插件據(jù)此推進(jìn)消費(fèi)者的虛擬隊(duì)列位置。也就是說插件其實(shí)是在 MongoDB 之上做了一層“事件流封裝”。你不需要自己去寫 tail oplog 的邏輯插件幫你搞定了。為了驗(yàn)證這個設(shè)計(jì)我在測試環(huán)境里往x-event交換器發(fā)了三條測試消息然后直接去 Mongo 里翻數(shù)據(jù)。文檔結(jié)構(gòu)大致如下{ _id: ObjectId(665e3a1f0f54f2aa1d63a101), exchange: event.log, routing_key: user.login, payload: {\uid\:10001,\action\:\login\}, timestamp: 2024-06-03T12:00:31.123Z, properties: { content_type: application/json } }注意這三個字段exchange是來源交換器名routing_key是路由鍵payload是消息體原文。這個存儲結(jié)構(gòu)讓重放變得特別自然——你想回放某天的事件直接用 MongoDB 查那個時(shí)間段的文檔重新投遞就行。2.3 與普通隊(duì)列的對比不是替代關(guān)系很多初學(xué)者會糾結(jié)“我到底該用普通隊(duì)列還是這個插件”。我的觀點(diǎn)是它們是兩種不同的東西別硬代替。普通 AMQP 隊(duì)列解決的是“任務(wù)分發(fā)”核心語義是消息被一個消費(fèi)者取走之后就屬于這個消費(fèi)者其他消費(fèi)者不會再拿到Work Queue 模式。這是點(diǎn)對點(diǎn)通信。而x-event交換器解決的是“事件廣播 存儲”核心語義是每個事件都會被持久化每個虛擬隊(duì)列都可以從頭拉取全量事件消費(fèi)者拿到的數(shù)據(jù)是“流式回放”不會因?yàn)閯e人消費(fèi)過就拿不到。這是類似 Kafka consumer group 的語義。語義對比直接看表對比維度普通 AMQP Queuex-event Event Exchange消費(fèi)后消息是否刪除是ack 后即刪除否長期保留在 Mongo多個消費(fèi)者能否獨(dú)立消費(fèi)全量否競爭消費(fèi)是各推各的游標(biāo)消息重放能力弱需要死信隊(duì)列等變通強(qiáng)隨時(shí)重新拉取消費(fèi)協(xié)議AMQP 協(xié)議HTTP API典型場景任務(wù)隊(duì)列、RPC、負(fù)載均衡事件溯源、審計(jì)日志、數(shù)據(jù)管道存儲依賴RabbitMQ 自身MongoDB所以如果你只想做“異步任務(wù)處理”老老實(shí)實(shí)用普通隊(duì)列別折騰這個插件。如果你需要的是“每個消費(fèi)者都能拿到全量事件且事件不丟可重放”x-event才是正解。3. 環(huán)境搭建與部署實(shí)操從安裝到跑通一條消息3.1 環(huán)境適配與準(zhǔn)備工作動手之前先確認(rèn)三件事RabbitMQ 版本、Erlang 版本、MongoDB 版本。rabbitmq-event-exchange插件不像普通插件那樣裝了就能用它對 MongoDB 的驅(qū)動和版本有隱性要求。我在第一次安裝時(shí)就栽過跟頭。當(dāng)時(shí)圖省事在一個老環(huán)境的 RabbitMQ 3.8.x 上直接啟用了插件結(jié)果轉(zhuǎn)發(fā)消息時(shí) Mongo 那邊一直報(bào) offset 相關(guān)錯誤后來查社區(qū)討論才知道老版本插件對 MongoDB 新版驅(qū)動兼容性一般。如果你手頭有選擇余地我的建議是RabbitMQ 版本3.8.x 或 3.9.x 都可以這兩個版本對插件支持最穩(wěn)定。RabbitMQ 3.13 之后的版本沒有繼續(xù)跟進(jìn)這個插件官方倉庫基本處于維護(hù)停滯狀態(tài)不建議新項(xiàng)目用 4.x 去賭兼容性。MongoDB 版本3.6 到 4.4 之間比較穩(wěn)。如果你用的是 MongoDB 5.0 或更高版本要特別留意 tailable cursor 的行為變化我后面會在問題排查部分詳細(xì)說。操作系統(tǒng)Linux 環(huán)境最省心。Windows 上跑 RabbitMQ 本身能跑但插件相關(guān)的本地調(diào)試有時(shí)候會遇到文件路徑和權(quán)限問題不建議新手用 Windows 練這個。三個服務(wù)都準(zhǔn)備好之后你還需要裝好 RabbitMQ 管理插件rabbitmq_management因?yàn)?Event Exchange 插件有一部分管理操作需要借助管理 HTTP API 完成且調(diào)試時(shí)能直接在后臺看到x-event交換器的狀態(tài)非常方便。3.2 插件安裝與啟用步驟這個插件在官方 rabbitmq-public-plugins 倉庫里能找到編譯好的發(fā)行包。操作分三步走第一步下載插件文件。這里可以直接從 GitHub 的 Releases 頁面下載對應(yīng) RabbitMQ 版本的.ez文件。下載后把文件放到 RabbitMQ 插件目錄。Linux 上常見路徑是/usr/lib/rabbitmq/lib/rabbitmq_server-3.9.x/plugins如果你用的是包管理器安裝的 RabbitMQ不確定插件目錄路徑可以通過rabbitmq-plugins directories命令查詢。第二步啟用插件。執(zhí)行rabbitmq-plugins enable rabbitmq_event_exchange注意插件名在命令行里用下劃線還是連字符不同文檔寫法不一致。我實(shí)測下來rabbitmq-plugins enable rabbitmq_event_exchange這種寫法可以正常識別。啟用后重啟 RabbitMQ 服務(wù)讓插件完全加載systemctl restart rabbitmq-server第三步驗(yàn)證插件是否加載成功?？床寮斜韗abbitmq-plugins list如果看到[E*] rabbitmq_event_exchange這樣的輸出說明插件已經(jīng)啟用。E*表示顯式啟用且生效。這里提醒一個很多人容易忽略的點(diǎn)插件的運(yùn)行依賴 MongoDB 客戶端連接而插件本身啟動時(shí)并不會主動去連 Mongo。只有當(dāng)你創(chuàng)建x-event類型的交換器、或調(diào)用相關(guān) HTTP API 時(shí)它才會去連接配置的 MongoDB 實(shí)例。所以插件啟動成功 ≠ 配置完成別在這一步放松警惕。3.3 關(guān)鍵配置讓插件知道往哪個 MongoDB 寫數(shù)據(jù)插件的 MongoDB 連接配置在 RabbitMQ 配置文件里。以新版 RabbitMQ 使用的rabbitmq.conf為例event_exchange.mongo_db mongodb://127.0.0.1:27017/events_dbevents_db是你自己指定的庫名插件會把事件文檔寫到這個庫下的集合里。如果 MongoDB 設(shè)置了用戶名密碼連接串要帶認(rèn)證信息event_exchange.mongo_db mongodb://user:pass127.0.0.1:27017/events_db?authSourceadmin配置完成后重啟 RabbitMQ。注意如果配置文件寫錯RabbitMQ 可能不會直接報(bào)錯而是等你真正創(chuàng)建x-event交換器時(shí)才在日志里報(bào)連接失敗。所以改完配置后一定要做一次“創(chuàng)建交換器 發(fā)消息”的聯(lián)調(diào)測試確保配置真的生效。配置完之后記得授權(quán) MongoDB 里對應(yīng)的用戶至少要有對events_db的讀寫權(quán)限。插件會創(chuàng)建 collection如果用戶權(quán)限不足消息轉(zhuǎn)發(fā)會靜默失敗看起來像是沒發(fā)出去。3.4 創(chuàng)建 x-event 交換器并驗(yàn)證消息落庫這一步可以直接用 RabbitMQ 管理后臺操作也可以用命令行工具。我習(xí)慣用命令行動手因?yàn)榉奖阌涗浐蛷?fù)現(xiàn)。用 rabbitmqadmin 創(chuàng)建一個名為event.log的交換器類型指定為x-eventrabbitmqadmin declare exchange nameevent.log typex-event durabletrue創(chuàng)建成功后往這個交換器發(fā)一條測試消息rabbitmqadmin publish exchangeevent.log routing_keyuser.login payload{uid:10001,action:login}然后立即去 MongoDB 里查看mongo --quiet events_db --eval db.event.log.find().sort({_id:-1}).limit(5).pretty()如果能看到類似前面第 2 章里的文檔結(jié)構(gòu)說明恭喜落庫成功說明 RabbitMQ 到 MongoDB 這一條鏈路已經(jīng)通了。注意x-event類型的交換器聲明為durabletrue有意義因?yàn)?RabbitMQ 重啟后插件會嘗試恢復(fù)交換器定義。實(shí)際測試中如果交換器不持久化重啟后綁定關(guān)系會丟失需要重新聲明。建議從第一天就養(yǎng)成durabletrue的習(xí)慣。4. 消費(fèi)端接入實(shí)操用 HTTP API 玩轉(zhuǎn)事件流4.1 創(chuàng)建虛擬綁定并拉取消息這個插件的消費(fèi)流程我拆成三步創(chuàng)建虛擬隊(duì)列綁定、輪詢拉取消息、更新游標(biāo)。用一個我們內(nèi)部的模擬場景來說明假設(shè)你現(xiàn)在要在另一個服務(wù)里消費(fèi)event.log里的全部登錄事件。第一步是創(chuàng)建虛擬隊(duì)列綁定。說白了就是通過 HTTP API 注冊一個“虛擬消費(fèi)者身份”告訴插件我要從event.log這個事件流消費(fèi)數(shù)據(jù)了。實(shí)際的調(diào)用格式在不同版本之間有一點(diǎn)差異但大體路子是一致的。調(diào)用 RabbitMQ 管理 API 創(chuàng)建一個綁定關(guān)系curl -X POST http://127.0.0.1:15672/api/bindings/event.log/event-queue \ -u guest:guest \ -H Content-Type: application/json \ -d {routing_key:user.login}這一步的語義就是把虛擬隊(duì)列event-queue綁定到event.log交換器上路由鍵為user.login。綁定關(guān)系創(chuàng)建后后續(xù)所有匹配的事件都可以通過這個虛擬隊(duì)列拉取。第二步通過插件暴露的 HTTP 消費(fèi)接口拉取消息。你可以用 curl 驗(yàn)證curl -X POST http://127.0.0.1:15672/api/queues/%2F/event-queue/get \ -u guest:guest \ -H Content-Type: application/json \ -d {count:10,ackmode:ack_requeue_true,encoding:auto}響應(yīng)的 JSON 數(shù)組里會包含消息內(nèi)容、路由鍵和唯一標(biāo)識等信息。第三步記錄返回消息里的游標(biāo) ID通常是eventId字段下一次請求時(shí)帶上它作為新的起點(diǎn)。這樣你的消費(fèi)者就有了“記憶”重啟服務(wù)也不會重復(fù)消費(fèi)已經(jīng)處理過的歷史事件。4.2 寫一個最小可用的 Node.js 消費(fèi)者光看 API 不直觀這里放一個我實(shí)際用過的 Node.js 消費(fèi)循環(huán)。邏輯很簡單不斷輪詢拉取事件解析游標(biāo)處理完畢后更新本地偏移量。const http require(http); const RABBIT_API http://127.0.0.1:15672; const USER guest; const PASS guest; const VHOST %2F; // 默認(rèn) vhost 是 /URL 編碼后就是 %2F const QUEUE event-queue; let cursor null; function fetchEvents() { const body JSON.stringify({ count: 50, ackmode: ack_requeue_true, encoding: auto, cursor: cursor || undefined }); const req http.request( { hostname: 127.0.0.1, port: 15672, path: /api/queues/${VHOST}/${QUEUE}/get, method: POST, auth: ${USER}:${PASS}, headers: { Content-Type: application/json } }, (res) { let data ; res.on(data, (chunk) (data chunk)); res.on(end, () { try { const events JSON.parse(data); for (const evt of events) { // 這里寫你的業(yè)務(wù)處理邏輯 console.log(handle event:, evt.routing_key, evt.payload); // 推進(jìn)游標(biāo) if (evt.eventId) cursor evt.eventId; } } catch (e) { console.error(parse error:, e.message); } }); } ); req.on(error, (e) console.error(request error:, e.message)); req.write(body); req.end(); } // 每 2 秒拉一次 setInterval(fetchEvents, 2000);這段代碼有幾個細(xì)節(jié)我特意寫成這樣都是有原因的ackmode用的是ack_requeue_true。因?yàn)橄⒋嬖?Mongo 里本身就不存在 AMQP 語義中的 ack/requeue。這樣寫主要是為了兼容管理 API 的字段校驗(yàn)實(shí)際上消息不會因?yàn)?ack 而被刪除。游標(biāo)變量在每次成功處理完一批事件后更新而不是在處理前更新。這是為了崩潰恢復(fù)時(shí)盡量不丟事件——寧可重復(fù)消費(fèi)不能跳事件。輪詢間隔建議按業(yè)務(wù)容忍延遲設(shè)置。2 秒是折中值如果延遲敏感可以縮短到 500ms但要注意請求頻率別把管理 API 打爆。4.3 多消費(fèi)者獨(dú)立推進(jìn)游標(biāo)的驗(yàn)證這個插件最吸引我的點(diǎn)就是多個消費(fèi)者可以各自維護(hù)游標(biāo)、互不干擾。我在測試環(huán)境里做了這樣一組實(shí)驗(yàn)消費(fèi)者 A 從event.log拉消息只處理user.login路由鍵的事件。消費(fèi)者 B 也從event.log拉消息但處理所有路由鍵的事件。我同時(shí)啟動了這兩個消費(fèi)進(jìn)程然后往交換器里連續(xù)發(fā)了 10 條不同路由鍵的消息。結(jié)果發(fā)現(xiàn)兩個消費(fèi)者各自都拿到了消息沒有出現(xiàn)“A 消費(fèi)了 B 就沒得消費(fèi)”的情況。原因就在于 A 和 B 各自維護(hù)著自己的 Mongo 游標(biāo)位置彼此隔離。這個能力落到業(yè)務(wù)上價(jià)值巨大。比如 A 服務(wù)可以做實(shí)時(shí)風(fēng)控B 服務(wù)可以把同一批事件同步到數(shù)據(jù)倉庫C 服務(wù)可以只統(tǒng)計(jì)運(yùn)營指標(biāo)。大家都讀同一份事件數(shù)據(jù)但各取所需誰都不影響誰。用普通隊(duì)列實(shí)現(xiàn)這套邏輯你得給每個服務(wù)單獨(dú)建隊(duì)列、單獨(dú)做綁定或者把消息復(fù)制多份麻煩得很。5. 常見問題與排查技巧實(shí)錄那些文檔里不會寫的事5.1 消息一直拉不到的排查路徑如果你按上面步驟做完創(chuàng)建了綁定、發(fā)了消息卻發(fā)現(xiàn) HTTP 接口返回空數(shù)組不要慌按順序排查下面幾項(xiàng)第一確認(rèn) Mongo 里有沒有數(shù)據(jù)。這是最基礎(chǔ)的一步。查不到數(shù)據(jù)就別糾結(jié)消費(fèi)端問題在寫入鏈路。mongo --quiet events_db --eval db[event.log].count()如果 count 為 0說明消息根本沒落庫。檢查 RabbitMQ 日志看有沒有 Mongo 連接失敗、認(rèn)證失敗、collection 權(quán)限不足等報(bào)錯。我在測試時(shí)就遇到過一次Mongo 用戶權(quán)限只給了庫的 read 權(quán)限沒給 write 權(quán)限RabbitMQ 日志里只飄了一條無關(guān)痛癢的 warning消息就默默丟了。第二確認(rèn)綁定關(guān)系用的是正確的交換器和路由鍵。x-event交換器在消息轉(zhuǎn)發(fā)時(shí)如果虛擬隊(duì)列的路由鍵與消息的 routing_key 不匹配消息雖然落庫了但虛擬隊(duì)列拉取時(shí)過濾不到。這里沒有 AMQP 那樣的死信提示排查起來全靠對路由鍵。建議先用 fanout 式的全匹配測試——把虛擬隊(duì)列的路由鍵設(shè)置成#或去掉路由鍵確認(rèn)能拉到全量消息再逐步縮小過濾范圍。第三檢查游標(biāo)參數(shù)是否傳正確。如果你消費(fèi)者在代碼里把游標(biāo)值寫死了而且這個游標(biāo)值已經(jīng)大于當(dāng)前最新文檔的_id結(jié)果自然就是拉不到任何消息。我踩過這個坑我把測試環(huán)境里的游標(biāo)值誤寫到了生產(chǎn)代碼里結(jié)果那個消費(fèi)者跑了一天也沒拉到數(shù)據(jù)。排查方式很簡單把游標(biāo)參數(shù)去掉再請求一次如果能拉到消息說明游標(biāo)推進(jìn)或初始化邏輯有問題。5.2 消息重復(fù)消費(fèi)問題這個插件的消費(fèi)語義是“游標(biāo)推進(jìn)式”不是 AMQP 的“ack 刪除式”。所以只要消費(fèi)者在處理完業(yè)務(wù)邏輯之前崩潰重啟后游標(biāo)沒有更新消息就會再次被拉到。這是設(shè)計(jì)使然不算 bug但你要在應(yīng)用層做好冪等。我的處理習(xí)慣是給每條消息的業(yè)務(wù)處理流程加上一個“處理中”標(biāo)記。最簡單的方式是在業(yè)務(wù)數(shù)據(jù)庫里建一張事件處理記錄表字段包括event_id、consumer_name、handle_time用event_id consumer_name做唯一索引。每次處理前先嘗試插入插入失敗說明這條消息已經(jīng)被處理過直接跳過。這個方案聽起來土但在消息量不大時(shí)非?？煽?。如果你用的是 MongoDB更簡單的辦法是直接利用事件文檔里的_id做冪等鍵。因?yàn)槭录?ID 本身全局唯一把業(yè)務(wù)更新和事件消費(fèi)記錄放在同一個 Mongo 事務(wù)里Mongo 4.0 之后支持多文檔事務(wù)從機(jī)制上保證不重復(fù)。不過要提醒一下事務(wù)有性能開銷適合審計(jì)類、交易類的強(qiáng)一致性場景不適合高吞吐日志流。5.3 性能與存儲膨脹問題很多人不敢用這個插件核心顧慮就是性能。畢竟發(fā)一條消息要寫一次 Mongo比普通內(nèi)存隊(duì)列慢多了。我在測試環(huán)境做過一次粗測單機(jī)部署、默認(rèn)配置下往x-event交換器持續(xù)發(fā)送 1KB 左右的消息穩(wěn)定吞吐大約在每秒兩千到三千條延遲取決于 Mongo 的寫入耗時(shí)。這個吞吐跑業(yè)務(wù)審計(jì)、事件收集完全夠用但如果你想用它做高并發(fā)消息中轉(zhuǎn)肯定不合適。存儲膨脹倒是更容易踩的坑。因?yàn)橄⑷柯鋷烨也磺宄辗e月累 Mongo 的數(shù)據(jù)量會非?？捎^。我做審計(jì)系統(tǒng)時(shí)跑了一個月光event.logcollection 就攢了幾個 GB。建議在生產(chǎn)環(huán)境里盡早規(guī)劃數(shù)據(jù)生命周期定期任務(wù)刪除過期數(shù)據(jù)比如只保留 30 天?；蛘哂?MongoDB 的 TTL 索引。但是注意TTL 索引是基于一個日期字段的而插件的文檔里沒有專門的過期時(shí)間字段你可以靠timestamp字段建索引MongoDB 的 TTL 線程會自動清理過期文檔。還有一招是分 collection 存儲按天動態(tài)切換 collection比如event.log.20240603、event.log.20240604到期后直接 drop 整個 collection。這個方案對磁盤回收最友好因?yàn)?MongoDB 刪除 collection 是實(shí)時(shí)釋放空間的而普通 delete 操作只是標(biāo)記刪除磁盤空間不會立刻回收。5.4 關(guān)于 MongoDB 新版本的兼容性忠告這一點(diǎn)寫出來可能有些人不愛看但確實(shí)是實(shí)際經(jīng)歷。社區(qū)里反饋這個插件在 MongoDB 3.2 到 4.4 版本區(qū)間內(nèi)表現(xiàn)最穩(wěn)定。到了 MongoDB 5.0 之后部分 tailable cursor 行為和驅(qū)動行為有調(diào)整插件如果沒有跟著更新可能會出現(xiàn)“拉取到歷史數(shù)據(jù)后游標(biāo)不推進(jìn)”“連接被 Mongo 主動斷開后不會重連”等怪問題。我在測試環(huán)境里用 MongoDB 5.0 跑過一次現(xiàn)象是消費(fèi)者第一次能正常拉取一批數(shù)據(jù)但游標(biāo)停在某個位置后再也不動了RabbitMQ 日志和 Mongo 日志都沒有明確報(bào)錯。折騰了很久才發(fā)現(xiàn)是新版驅(qū)動默認(rèn)對 cursor 超時(shí)機(jī)制更激進(jìn)插件又沒有主動設(shè)置noCursorTimeout導(dǎo)致游標(biāo)被服務(wù)端回收了。如果你必須使用 MongoDB 5.0 及以上版本我的建議是先在測試環(huán)境把消費(fèi)流程完整跑 24 小時(shí)確認(rèn)游標(biāo)能持續(xù)推進(jìn)、不會中途卡死再部署生產(chǎn)。不要在沒驗(yàn)證的情況下直接升級這種冷門插件的坑一旦觸發(fā)排查代價(jià)很高。6. 最后再分享一點(diǎn)我的實(shí)操心得文章寫到這里主體內(nèi)容基本講完了但我還是想用最后一段說點(diǎn)掏心窩的話。這個插件不是 RabbitMQ 生態(tài)里最耀眼的存在它甚至有點(diǎn)“老派”。它火的年代Kafka 還不像今天這樣普及很多團(tuán)隊(duì)手里只有 RabbitMQ 和 MongoDB卻想實(shí)現(xiàn)事件流、事件溯源這類需求于是官方就做了這個橋接方案。放到今天來看它的設(shè)計(jì)思路依然讓人拍案叫絕——用 ObjectId 做游標(biāo)、用 tailable cursor 做流式拉取、用 HTTP 做消費(fèi)協(xié)議每一步都踩在 MongoDB 的特性上實(shí)現(xiàn)得非常優(yōu)雅。但從維護(hù)角度看這個插件畢竟不是一個高頻更新的項(xiàng)目。如果你要在嚴(yán)肅的生產(chǎn)環(huán)境里長期依賴它一定要做好備份和監(jiān)控尤其是 Mongo 連接狀態(tài)、游標(biāo)推進(jìn)情況、存儲增長趨勢這三項(xiàng)指標(biāo)。我自己的建議是給它加上獨(dú)立的健康檢查任務(wù)定期通過 API 拉一條測試消息并消費(fèi)確保鏈路隨時(shí)可用。如果你現(xiàn)在的項(xiàng)目規(guī)模不大、事件量沒到每秒幾萬條的程度同時(shí)你又需要事件可重放、多消費(fèi)者獨(dú)立推進(jìn)那么這個插件真的值得一試。拿它做完審計(jì)日志或事件總線的底座你會發(fā)現(xiàn)自己省掉了不少重復(fù)造輪子的時(shí)間。如果你有別的落地場景也歡迎在評論區(qū)聊聊我很想看看大家是怎么玩這個寶藏插件的。