MongoDB Change Stream初體驗

1fe1se 發(fā)布于2019-06-26 18:04 / 3575人閱讀

摘要：可以發(fā)現(xiàn)，整個同步過程是依賴于來進行的。不考慮導致的問題，正常的應用升級也會導致應用中斷運行。注意事項為了避免被回滾的更新被發(fā)布出去，選擇只在一個變更到達大多數(shù)節(jié)點不可能被回滾時，才會將這些變更發(fā)布到應用。

Change Stream是MongoDB從3.6開始支持的新特性。這個新特性有哪些奇妙之處，會給我們帶來什么便利？本次的文章將就這個主題進行初步討論。

Change Stream是什么？

顧名思義，Change Stream即變更流，是MongoDB向應用發(fā)布數(shù)據(jù)變更的一種方式。即當數(shù)據(jù)庫中有任何數(shù)據(jù)發(fā)生變化，應用端都可以得到通知。我們可以將其理解為在應用中執(zhí)行的觸發(fā)器。至于應用想得到什么數(shù)據(jù)，以什么形式得到數(shù)據(jù)，則可以通過聚合框架加以過濾和轉(zhuǎn)換。這點將在后文中討論。

Change Stream的原理

我們先來回顧一下MongoDB復制集大致是如何工作的：

應用通過驅(qū)動向數(shù)據(jù)庫發(fā)起寫入請求；

在同一個事務中，MongoDB完成oplog和集合的修改；

oplog被其他從節(jié)點拉走；

從節(jié)點應用得到的oplog，同樣在一個事務中完成對oplog和集合的修改；

至此，復制集同步完成。可以發(fā)現(xiàn)，整個同步過程是依賴于oplog來進行的。也就是說oplog實際上已經(jīng)包含了我們需要的所有變更數(shù)據(jù)。如果觀測oplog的變化，是否就能夠得到所有變更的數(shù)據(jù)了呢？對，change stream正是基于這個原理實現(xiàn)的。但事情并沒有這么簡單！我們來看一下問題有可能出在什么地方。

如何從斷點恢復

現(xiàn)實世界中，沒有哪個應用是可以不間斷運行的。不考慮bug導致的問題，正常的應用升級也會導致應用中斷運行。那么在應用恢復的時候，從哪里開始繼續(xù)獲取變更呢？oplog當然是可以幫我們做到這點的，但你必須對MongoDB足夠了解，才知道有oplogReplay這樣的參數(shù)，以及其他一些問題。

如何有效地處理訂閱

假設在一個應用中需要訂閱10個不同集合的變更情況，是否需要開10個tailable cursor去獲取oplog的變更呢？如果是100個集合呢？出于效率考慮顯然不應該這么做。那么整個過程就會變成一個生產(chǎn)者-消費者模式，由一個線程負責從oplog獲取變更，由訂閱的線程負責消費這些變更。雖然實現(xiàn)也不是那么復雜，并且多半可以找到開源實現(xiàn)，但是涉及多線程就已經(jīng)足夠讓初學者頭疼一陣的了。
公平地說，上面這些還不算嚴重的問題，下面這些問題可能會更讓人頭疼。

如何管理權(quán)限

想要tail oplog，必須對local.oplog.rs有讀權(quán)限。實際上這相當于對整個數(shù)據(jù)庫都有了讀權(quán)限，因為所有的變更都會在這里體現(xiàn)出來。DBA可能會阻止你這么做，因為這實在不是一個很安全的做法。

如何數(shù)據(jù)回滾

極端情況下，如果應用處理不當，MongoDB中可能發(fā)生數(shù)據(jù)回滾rollback的問題。如果僅僅通過跟蹤oplog，則會出現(xiàn)已經(jīng)通知出去的變更被回滾的情況。

幸運的是上面這些問題現(xiàn)在都不是問題了，因為change stream幫我們規(guī)避了這些復雜的細節(jié)。

使用方法

由于各種驅(qū)動都會有不同的語法和API，從shell中嘗試使用change stream可能是最簡便的方法。這并不妨礙你隨后在各種驅(qū)動中的使用，因為shell中能實現(xiàn)的功能在驅(qū)動中一定有對應的語法。下面就以shell為例看看change stream應該如何使用。

打開一個shell，訂閱你需要關(guān)注的集合
比如：

var cursor = db.bar.watch();

為了便于演示，我們在這個shell中不斷遍歷這個游標以獲取新數(shù)據(jù)：

while(true) {
    if (cursor.hasNext()) {
        print(JSON.stringify(cursor.next()));
    }
}

打開另一個shell，向bar集合中插入一條數(shù)據(jù)：

db.bar.insert({y: 1})

此時第一個shell中會立即輸出變更數(shù)據(jù)：

{"_id":{"_data":{"$binary":"glzquiIAAAACRmRfaWQAZFzquiK0lDNo+K0DpwBaEARUMrm0ruVACoftuxjt1RtCBA==","$type":"00"}},"operationType":"insert","fullDocument":{"_id":{"$oid":"5ceaba22b4943368f8ad03a7"},"y":1},"ns":{"db":"test","coll":"bar"},"documentKey":{"_id":{"$oid":"5ceaba22b4943368f8ad03a7"}}}

這里的一些字段的簡單介紹。更完整的介紹請查閱文檔change events：

_id: 用于恢復斷點時使用。即知道這個值，應用斷開后下次重啟里就可以從這個斷點之后開始恢復獲得變更；

operationType: 操作類型，常見的值包括：

insert

update

delete

ns: 正在操作的命名空間

fullDocument: 完整的文檔

從斷點恢復

var cursor = db.bar.watch([], {resumeAfter: <\_id>})

此時使用hasNext()/next()即可獲取到隨后的變更。

注意事項 {readConcern: "majority"}

為了避免被回滾的更新被發(fā)布出去，change stream選擇只在一個變更到達大多數(shù)節(jié)點（不可能被回滾）時，才會將這些變更發(fā)布到應用。使用的方式即{readConcern: "majority"}。因此以下這些情況下change stream都是不會向應用通知任何變更的：

禁用了readConcern；

從舊版本升級，但沒有更新featureCompatibilityVersion；

PSA架構(gòu)中S宕機；

斷點可恢復時間

因為change stream是依賴于oplog工作的，自然也會面臨oplog面臨的所有問題。問題之一就是oplog被覆蓋。因此想要保證斷點可以恢復，必須保證應用在oplog window的時間內(nèi)請求斷點。

刪除集合

如果在訂閱集合變更過程中集合被刪除，則會收到一條invalid信息通知，表示集合已不再可用：

{
    "_id" : {
        "_data" : BinData(0,"glzqxCcAAAACFFoQBFQyubSu5UAKh+27GO3VG0IE")
    },
    "operationType" : "invalidate"
}

參考資料

Tailable cursor: https://docs.mongodb.com/manual/core/tailable-cursors/

生產(chǎn)者-消費者模式: https://zh.wikipedia.org/wiki/%E7%94%9F%E4%BA%A7%E8%80%85%E6%B6%88%E8%B4%B9%E8%80%85%E9%97%AE%E9%A2%98

關(guān)于回滾: https://docs.mongodb.com/manual/core/replica-set-rollbacks/

變更事件: https://docs.mongodb.com/manual/reference/change-events/

Change Stream介紹文檔：https://docs.mongodb.com/manual/changeStreams/

作者簡介

張耀星，MongoDB亞太區(qū)首席技術(shù)咨詢服務顧問。在MongoDB的開發(fā)、應用和咨詢服務上有多年實踐經(jīng)驗。作為MongoDB認證專家，曾經(jīng)為不同行業(yè)的各類大型客戶提供過培訓、性能調(diào)優(yōu)、架構(gòu)設計等各類MongoDB相關(guān)技術(shù)服務。