不想引戰(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í)需求。某段時間我在做一個內(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 官方插件列表時看到rabbitmq-event-exchange才真正找到順手的方案。這個插件的定位很明確它為 RabbitMQ 增加了一種新的交換器類型x-event。當(dāng)你把消息發(fā)到x-event類型的交換器上時消息并不會像普通交換器那樣被路由到某個 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)位置。同一個事件流集合可以同時存在多個虛擬隊(duì)列每個虛擬隊(duì)列各自獨(dú)立推進(jìn)游標(biāo)互不影響。實(shí)際操作中建立綁定關(guān)系的姿勢和普通 RabbitMQ 不太一樣——你不太需要去聲明一個普通 AMQP 隊(duì)列然后 bind 到 exchange 上當(dāng)然插件也兼容了部分 AMQP 操作更多時候是通過插件自帶的 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 自帶時間戳且單調(diào)遞增同一秒內(nèi)順序也遞增天然適合作為事件游標(biāo)。消費(fèi)者通過插件 HTTP API 拉取事件時插件會在 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 查那個時間段的文檔重新投遞就行。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)隨時重新拉取消費(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ū)動和版本有隱性要求。我在第一次安裝時就栽過跟頭。當(dāng)時圖省事在一個老環(huán)境的 RabbitMQ 3.8.x 上直接啟用了插件結(jié)果轉(zhuǎn)發(fā)消息時 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)試有時候會遇到文件路徑和權(quán)限問題不建議新手用 Windows 練這個。三個服務(wù)都準(zhǔn)備好之后你還需要裝好 RabbitMQ 管理插件rabbitmq_management因?yàn)?Event Exchange 插件有一部分管理操作需要借助管理 HTTP API 完成且調(diào)試時能直接在后臺看到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 客戶端連接而插件本身啟動時并不會主動去連 Mongo。只有當(dāng)你創(chuàng)建x-event類型的交換器、或調(diào)用相關(guān) HTTP API 時它才會去連接配置的 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交換器時才在日志里報(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字段下一次請求時帶上它作為新的起點(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ù)時盡量不丟事件——寧可重復(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拉消息但處理所有路由鍵的事件。我同時啟動了這兩個消費(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ù)上價值巨大。比如 A 服務(wù)可以做實(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)錯。我在測試時就遇到過一次Mongo 用戶權(quán)限只給了庫的 read 權(quán)限沒給 write 權(quán)限RabbitMQ 日志里只飄了一條無關(guān)痛癢的 warning消息就默默丟了。第二確認(rèn)綁定關(guān)系用的是正確的交換器和路由鍵。x-event交換器在消息轉(zhuǎn)發(fā)時如果虛擬隊(duì)列的路由鍵與消息的 routing_key 不匹配消息雖然落庫了但虛擬隊(duì)列拉取時過濾不到。這里沒有 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)被處理過直接跳過。這個方案聽起來土但在消息量不大時非?？煽?。如果你用的是 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 的寫入耗時。這個吞吐跑業(yè)務(wù)審計(jì)、事件收集完全夠用但如果你想用它做高并發(fā)消息中轉(zhuǎn)肯定不合適。存儲膨脹倒是更容易踩的坑。因?yàn)橄⑷柯鋷烨也磺宄辗e月累 Mongo 的數(shù)據(jù)量會非?？捎^。我做審計(jì)系統(tǒng)時跑了一個月光event.logcollection 就攢了幾個 GB。建議在生產(chǎn)環(huán)境里盡早規(guī)劃數(shù)據(jù)生命周期定期任務(wù)刪除過期數(shù)據(jù)比如只保留 30 天?；蛘哂?MongoDB 的 TTL 索引。但是注意TTL 索引是基于一個日期字段的而插件的文檔里沒有專門的過期時間字段你可以靠timestamp字段建索引MongoDB 的 TTL 線程會自動清理過期文檔。還有一招是分 collection 存儲按天動態(tài)切換 collection比如event.log.20240603、event.log.20240604到期后直接 drop 整個 collection。這個方案對磁盤回收最友好因?yàn)?MongoDB 刪除 collection 是實(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 超時機(jī)制更激進(jìn)插件又沒有主動設(shè)置noCursorTimeout導(dǎo)致游標(biāo)被服務(wù)端回收了。如果你必須使用 MongoDB 5.0 及以上版本我的建議是先在測試環(huán)境把消費(fèi)流程完整跑 24 小時確認(rèn)游標(biāo)能持續(xù)推進(jìn)、不會中途卡死再部署生產(chǎn)。不要在沒驗(yàn)證的情況下直接升級這種冷門插件的坑一旦觸發(fā)排查代價很高。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)確保鏈路隨時可用。如果你現(xiàn)在的項(xiàng)目規(guī)模不大、事件量沒到每秒幾萬條的程度同時你又需要事件可重放、多消費(fèi)者獨(dú)立推進(jìn)那么這個插件真的值得一試。拿它做完審計(jì)日志或事件總線的底座你會發(fā)現(xiàn)自己省掉了不少重復(fù)造輪子的時間。如果你有別的落地場景也歡迎在評論區(qū)聊聊我很想看看大家是怎么玩這個寶藏插件的。 RabbitMQ Event Exchange：消息落庫MongoDB的冷門寶藏插件
不想引戰(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í)需求。某段時間我在做一個內(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 官方插件列表時看到rabbitmq-event-exchange才真正找到順手的方案。這個插件的定位很明確它為 RabbitMQ 增加了一種新的交換器類型x-event。當(dāng)你把消息發(fā)到x-event類型的交換器上時消息并不會像普通交換器那樣被路由到某個 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)位置。同一個事件流集合可以同時存在多個虛擬隊(duì)列每個虛擬隊(duì)列各自獨(dú)立推進(jìn)游標(biāo)互不影響。實(shí)際操作中建立綁定關(guān)系的姿勢和普通 RabbitMQ 不太一樣——你不太需要去聲明一個普通 AMQP 隊(duì)列然后 bind 到 exchange 上當(dāng)然插件也兼容了部分 AMQP 操作更多時候是通過插件自帶的 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 自帶時間戳且單調(diào)遞增同一秒內(nèi)順序也遞增天然適合作為事件游標(biāo)。消費(fèi)者通過插件 HTTP API 拉取事件時插件會在 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 查那個時間段的文檔重新投遞就行。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)隨時重新拉取消費(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ū)動和版本有隱性要求。我在第一次安裝時就栽過跟頭。當(dāng)時圖省事在一個老環(huán)境的 RabbitMQ 3.8.x 上直接啟用了插件結(jié)果轉(zhuǎn)發(fā)消息時 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)試有時候會遇到文件路徑和權(quán)限問題不建議新手用 Windows 練這個。三個服務(wù)都準(zhǔn)備好之后你還需要裝好 RabbitMQ 管理插件rabbitmq_management因?yàn)?Event Exchange 插件有一部分管理操作需要借助管理 HTTP API 完成且調(diào)試時能直接在后臺看到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 客戶端連接而插件本身啟動時并不會主動去連 Mongo。只有當(dāng)你創(chuàng)建x-event類型的交換器、或調(diào)用相關(guān) HTTP API 時它才會去連接配置的 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交換器時才在日志里報(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字段下一次請求時帶上它作為新的起點(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ù)時盡量不丟事件——寧可重復(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拉消息但處理所有路由鍵的事件。我同時啟動了這兩個消費(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ù)上價值巨大。比如 A 服務(wù)可以做實(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)錯。我在測試時就遇到過一次Mongo 用戶權(quán)限只給了庫的 read 權(quán)限沒給 write 權(quán)限RabbitMQ 日志里只飄了一條無關(guān)痛癢的 warning消息就默默丟了。第二確認(rèn)綁定關(guān)系用的是正確的交換器和路由鍵。x-event交換器在消息轉(zhuǎn)發(fā)時如果虛擬隊(duì)列的路由鍵與消息的 routing_key 不匹配消息雖然落庫了但虛擬隊(duì)列拉取時過濾不到。這里沒有 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)被處理過直接跳過。這個方案聽起來土但在消息量不大時非?？煽?。如果你用的是 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 的寫入耗時。這個吞吐跑業(yè)務(wù)審計(jì)、事件收集完全夠用但如果你想用它做高并發(fā)消息中轉(zhuǎn)肯定不合適。存儲膨脹倒是更容易踩的坑。因?yàn)橄⑷柯鋷烨也磺宄辗e月累 Mongo 的數(shù)據(jù)量會非?？捎^。我做審計(jì)系統(tǒng)時跑了一個月光event.logcollection 就攢了幾個 GB。建議在生產(chǎn)環(huán)境里盡早規(guī)劃數(shù)據(jù)生命周期定期任務(wù)刪除過期數(shù)據(jù)比如只保留 30 天?；蛘哂?MongoDB 的 TTL 索引。但是注意TTL 索引是基于一個日期字段的而插件的文檔里沒有專門的過期時間字段你可以靠timestamp字段建索引MongoDB 的 TTL 線程會自動清理過期文檔。還有一招是分 collection 存儲按天動態(tài)切換 collection比如event.log.20240603、event.log.20240604到期后直接 drop 整個 collection。這個方案對磁盤回收最友好因?yàn)?MongoDB 刪除 collection 是實(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 超時機(jī)制更激進(jìn)插件又沒有主動設(shè)置noCursorTimeout導(dǎo)致游標(biāo)被服務(wù)端回收了。如果你必須使用 MongoDB 5.0 及以上版本我的建議是先在測試環(huán)境把消費(fèi)流程完整跑 24 小時確認(rèn)游標(biāo)能持續(xù)推進(jìn)、不會中途卡死再部署生產(chǎn)。不要在沒驗(yàn)證的情況下直接升級這種冷門插件的坑一旦觸發(fā)排查代價很高。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)確保鏈路隨時可用。如果你現(xiàn)在的項(xiàng)目規(guī)模不大、事件量沒到每秒幾萬條的程度同時你又需要事件可重放、多消費(fèi)者獨(dú)立推進(jìn)那么這個插件真的值得一試。拿它做完審計(jì)日志或事件總線的底座你會發(fā)現(xiàn)自己省掉了不少重復(fù)造輪子的時間。如果你有別的落地場景也歡迎在評論區(qū)聊聊我很想看看大家是怎么玩這個寶藏插件的。