Change Streams 是 MongoDB 提供的一種監(jiān)聽(tīng)數(shù)據(jù)變化的機(jī)制，基于復(fù)制集或分片集群的日志（oplog），以事件形式推送插入、更新、刪除等變更信息。1. 它適用于實(shí)時(shí)儀表盤(pán)、消息推送、同步服務(wù)等場(chǎng)景；2. 使用方式包括監(jiān)聽(tīng)單個(gè)集合、通過(guò)聚合管道過(guò)濾事件；3. 支持?jǐn)嗑€重連和恢復(fù)，確保不丟失事件；4. 常見(jiàn)應(yīng)用有數(shù)據(jù)同步、消息隊(duì)列替代、實(shí)時(shí) UI 更新、數(shù)據(jù)審計(jì)等；5. 實(shí)際使用中需注意性能影響、內(nèi)存占用、權(quán)限配置及分片支持版本要求。

Change Streams 是 MongoDB 提供的一種機(jī)制，用于監(jiān)聽(tīng)數(shù)據(jù)庫(kù)、集合或整個(gè)部署中的數(shù)據(jù)變化。它允許應(yīng)用程序在數(shù)據(jù)變更（如插入、更新、刪除）發(fā)生時(shí)立即收到通知，非常適合構(gòu)建需要實(shí)時(shí)響應(yīng)的系統(tǒng)。

如果你的應(yīng)用場(chǎng)景涉及實(shí)時(shí)儀表盤(pán)、消息推送、同步服務(wù)等，那 Change Streams 就非常實(shí)用。

什么是 Change Streams？

簡(jiǎn)單來(lái)說(shuō)，Change Streams 就是 MongoDB 中一種“監(jiān)聽(tīng)變化”的接口。它基于復(fù)制集或分片集群的日志（oplog），將每次數(shù)據(jù)變更以事件的形式推送給客戶端。

• 它不需要額外安裝插件，只要你的 MongoDB 支持副本集（replica set）或者分片集群，就可以直接使用。
• 每個(gè) change event 包含了操作類(lèi)型（insert, update, delete 等）、變更前后的內(nèi)容、時(shí)間戳等信息。
• 可以針對(duì)整個(gè)集群、某個(gè)數(shù)據(jù)庫(kù)、某張表，甚至可以通過(guò)聚合管道做過(guò)濾。

舉個(gè)例子：當(dāng)你在一個(gè)聊天應(yīng)用中添加了一條新消息，另一個(gè)用戶就能立刻看到這條消息——這就是 Change Streams 在背后發(fā)揮作用的典型場(chǎng)景。

如何使用 Change Streams 實(shí)現(xiàn)實(shí)時(shí)響應(yīng)？

要使用 Change Streams，你需要先建立一個(gè)監(jiān)聽(tīng)器，然后處理接收到的事件流。以下是幾個(gè)常見(jiàn)做法：

監(jiān)聽(tīng)單個(gè)集合的變化

const collection = db.collection('messages');
const changeStream = collection.watch();

changeStream.on('change', (change) => {
  console.log('檢測(cè)到變化:', change);
});

這種方式適用于你只關(guān)心特定集合的情況，比如訂單狀態(tài)變化、用戶行為日志等。

使用過(guò)濾器縮小關(guān)注范圍

你可以通過(guò)聚合管道來(lái)過(guò)濾你關(guān)心的事件。例如，只監(jiān)聽(tīng)更新操作：

const changeStream = collection.watch([
  { $match: { operationType: 'update' } }
]);

也可以結(jié)合 fullDocument 和 updateDescription 來(lái)獲取更詳細(xì)的變更內(nèi)容。

處理斷線重連和恢復(fù)

Change Streams 支持自動(dòng)從上次斷開(kāi)的位置繼續(xù)監(jiān)聽(tīng)，前提是你要開(kāi)啟 resumeAfter 或 startAtOperationTime：

const resumeToken = ...; // 上次保存的 token
const changeStream = collection.watch([], { resumeAfter: resumeToken });

這樣即使連接中斷，也不會(huì)丟失變更事件。

常見(jiàn)應(yīng)用場(chǎng)景有哪些？

實(shí)時(shí)數(shù)據(jù)同步

Change Streams 可以用來(lái)將 MongoDB 的變化同步到其他存儲(chǔ)系統(tǒng)，比如 Elasticsearch、Redis 或者另一個(gè)數(shù)據(jù)庫(kù)。每當(dāng)有新的文檔寫(xiě)入 MongoDB，就觸發(fā)同步邏輯，保持外部系統(tǒng)的數(shù)據(jù)一致性。

消息隊(duì)列替代方案

在一些輕量級(jí)系統(tǒng)中，Change Streams 可以作為簡(jiǎn)單的事件驅(qū)動(dòng)架構(gòu)的基礎(chǔ)。比如訂單創(chuàng)建后，自動(dòng)觸發(fā)一個(gè)異步任務(wù)去處理后續(xù)流程。

實(shí)時(shí) UI 更新

前端應(yīng)用可以通過(guò) WebSocket 接收來(lái)自后端的 Change Stream 事件，實(shí)現(xiàn)頁(yè)面的即時(shí)刷新，比如股票行情、聊天窗口、在線協(xié)作編輯等場(chǎng)景。

數(shù)據(jù)審計(jì)與監(jiān)控

記錄所有變更事件，用于審計(jì)或分析用途。你可以把每個(gè)變更都存入日志系統(tǒng)，方便排查問(wèn)題或生成報(bào)表。

用起來(lái)不復(fù)雜但容易忽略的地方

雖然 Change Streams 很強(qiáng)大，但在實(shí)際使用中也有一些需要注意的地方：

• 性能影響：頻繁的變更事件會(huì)增加網(wǎng)絡(luò)和 CPU 負(fù)擔(dān)，特別是在大規(guī)模寫(xiě)入的場(chǎng)景下。
• 內(nèi)存占用：如果事件處理慢于變更速度，可能會(huì)導(dǎo)致事件堆積，進(jìn)而影響內(nèi)存。
• 權(quán)限配置：確保運(yùn)行 Change Streams 的用戶有足夠的權(quán)限訪問(wèn) oplog。
• 分片支持：在分片集群中使用 Change Streams 需要 MongoDB 4.0 并啟用副本集。

總的來(lái)說(shuō)，Change Streams 是一個(gè)很適合做實(shí)時(shí)響應(yīng)的功能，使用得當(dāng)可以大大提升系統(tǒng)的反應(yīng)速度和用戶體驗(yàn)?；旧暇瓦@些。

以上是什么是變更流，如何用于對(duì)實(shí)時(shí)數(shù)據(jù)更改的反應(yīng)？的詳細(xì)內(nèi)容。更多信息請(qǐng)關(guān)注PHP中文網(wǎng)其他相關(guān)文章！