MongoDB Change Stream初體驗(yàn)

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

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

Change Stream是MongoDB從3.6開(kāi)始支持的新特性。這個(gè)新特性有哪些奇妙之處，會(huì)給我們帶來(lái)什么便利？本次的文章將就這個(gè)主題進(jìn)行初步討論。

Change Stream是什么？

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

Change Stream的原理

我們先來(lái)回顧一下MongoDB復(fù)制集大致是如何工作的：

應(yīng)用通過(guò)驅(qū)動(dòng)向數(shù)據(jù)庫(kù)發(fā)起寫(xiě)入請(qǐng)求；

在同一個(gè)事務(wù)中，MongoDB完成oplog和集合的修改；

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

從節(jié)點(diǎn)應(yīng)用得到的oplog，同樣在一個(gè)事務(wù)中完成對(duì)oplog和集合的修改；

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

如何從斷點(diǎn)恢復(fù)

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

如何有效地處理訂閱

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

如何管理權(quán)限

想要tail oplog，必須對(duì)local.oplog.rs有讀權(quán)限。實(shí)際上這相當(dāng)于對(duì)整個(gè)數(shù)據(jù)庫(kù)都有了讀權(quán)限，因?yàn)樗械淖兏紩?huì)在這里體現(xiàn)出來(lái)。DBA可能會(huì)阻止你這么做，因?yàn)檫@實(shí)在不是一個(gè)很安全的做法。

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

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

幸運(yùn)的是上面這些問(wèn)題現(xiàn)在都不是問(wèn)題了，因?yàn)閏hange stream幫我們規(guī)避了這些復(fù)雜的細(xì)節(jié)。

使用方法

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

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

var cursor = db.bar.watch();

為了便于演示，我們?cè)谶@個(gè)shell中不斷遍歷這個(gè)游標(biāo)以獲取新數(shù)據(jù)：

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

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

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

此時(shí)第一個(gè)shell中會(huì)立即輸出變更數(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"}}}

這里的一些字段的簡(jiǎn)單介紹。更完整的介紹請(qǐng)查閱文檔change events：

_id: 用于恢復(fù)斷點(diǎn)時(shí)使用。即知道這個(gè)值，應(yīng)用斷開(kāi)后下次重啟里就可以從這個(gè)斷點(diǎn)之后開(kāi)始恢復(fù)獲得變更；

operationType: 操作類(lèi)型，常見(jiàn)的值包括：

insert

update

delete

ns: 正在操作的命名空間

fullDocument: 完整的文檔

從斷點(diǎn)恢復(fù)

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

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

注意事項(xiàng) {readConcern: "majority"}

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

禁用了readConcern；

從舊版本升級(jí)，但沒(méi)有更新featureCompatibilityVersion；

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

斷點(diǎn)可恢復(fù)時(shí)間

因?yàn)閏hange stream是依賴(lài)于oplog工作的，自然也會(huì)面臨oplog面臨的所有問(wèn)題。問(wèn)題之一就是oplog被覆蓋。因此想要保證斷點(diǎn)可以恢復(fù)，必須保證應(yīng)用在oplog window的時(shí)間內(nèi)請(qǐng)求斷點(diǎn)。

刪除集合

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

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

參考資料

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

生產(chǎn)者-消費(fèi)者模式: 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/

作者簡(jiǎn)介

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