[[425468]]

簡(jiǎn)介：基于MongoDB的應(yīng)用程序通過(guò)Change Streams功能可以方便的實(shí)現(xiàn)對(duì)某個(gè)集合，數(shù)據(jù)庫(kù)或者整個(gè)集群的數(shù)據(jù)變更的訂閱，極大的方便了應(yīng)用對(duì)數(shù)據(jù)庫(kù)變化的感知，但是當(dāng)前Change Streams對(duì)部分?jǐn)?shù)據(jù)的變化并沒(méi)有提供對(duì)應(yīng)的事件（創(chuàng)建索引，刪除索引，shardCollection）等，本文介紹一種新的事件訂閱方式，來(lái)完善上述不足，并探討通過(guò)并發(fā)預(yù)讀的方式，來(lái)提升原生Change Streams的性能。

一、前言

MongoDB作為一款優(yōu)秀的NOSQL數(shù)據(jù)庫(kù)，支持海量存儲(chǔ)，查詢能力豐富以及優(yōu)秀的性能和可靠性，當(dāng)前大部分云廠商都提供了兼容MongoDB協(xié)議的服務(wù)，用戶使用廣泛，深受國(guó)內(nèi)外用戶和企業(yè)的認(rèn)可。

MongoDB從3.6版本開(kāi)始提供了Change Stream特性，通過(guò)該特性，應(yīng)用程序可以實(shí)時(shí)的訂閱特定集合、庫(kù)、或整個(gè)集群的數(shù)據(jù)變更事件，相比該特性推出之前通過(guò)監(jiān)聽(tīng)oplog的變化來(lái)實(shí)現(xiàn)對(duì)數(shù)據(jù)變更的感知，非常的易用，該特性同時(shí)支持副本集和集群場(chǎng)景。

Change Streams功能目前支持大部分?jǐn)?shù)據(jù)操作的事件，但是對(duì)于與部分其他操作，如創(chuàng)建索引，刪除索引，ColMod, shardCollection并不支持，而且目前Change Streams內(nèi)部實(shí)現(xiàn)是通過(guò)Aggregate命令的方式完成的， 對(duì)于分片集群場(chǎng)景下， 在mongos節(jié)點(diǎn)是通過(guò)單線程匯聚的方式完成從shard節(jié)點(diǎn)上oplog的拉取和處理，當(dāng)實(shí)例寫入壓力很大的情況下，感知數(shù)據(jù)的實(shí)時(shí)變化會(huì)有延遲，性能有待提升，對(duì)于ChangeStreams目前的性能問(wèn)題，官方也有過(guò)探討https://jira.mongodb.org/browse/SERVER-46979。

本文通過(guò)深入分析當(dāng)前的Change Stream實(shí)現(xiàn)機(jī)制，結(jié)合客戶實(shí)際使用場(chǎng)景，提出了一種新的多并發(fā)預(yù)讀的事件監(jiān)聽(tīng)方式，來(lái)解決上述問(wèn)題，并應(yīng)用到客戶實(shí)際遷移和數(shù)據(jù)庫(kù)容災(zāi)的場(chǎng)景中。

二、Change Steams 機(jī)制介紹

Change Streams支持對(duì)單個(gè)集合，DB，集群進(jìn)行事件訂閱，當(dāng)業(yè)務(wù)程序通過(guò)watch的方式發(fā)起訂閱后，背后發(fā)生了什么，讓我們一起來(lái)分析一下。

Change Streams內(nèi)部實(shí)現(xiàn)是通過(guò)Aggregate的方式實(shí)現(xiàn)的，所以watch背后，對(duì)應(yīng)的是客戶端向MongoDB Server發(fā)起了一個(gè)Aggregate命令，且對(duì)Aggregate的pipeline 參數(shù)中，添加了一個(gè)$changeStream的Stage, 結(jié)合客戶端其他參數(shù)，一起發(fā)給MongoDB Server。

當(dāng)Mongo Server收到Aggregate命令后，解析后，會(huì)根據(jù)具體的請(qǐng)求，組合一個(gè)新的Aggregate命令，并把該命令發(fā)給對(duì)應(yīng)的Shard節(jié)點(diǎn)，同時(shí)會(huì)在游標(biāo)管理器（CursorManger）中注冊(cè)一個(gè)新的游標(biāo)（cursor），并把游標(biāo)Id返回給客戶端。

當(dāng)Shard Server端收到Aggregate命令后，構(gòu)建pipeline流水線，并根據(jù)pipeline參數(shù)中包括了Change Steams參數(shù)，確定原始掃描的集合為oplog，并創(chuàng)建對(duì)該集合上掃描數(shù)據(jù)的原始cursor, 和對(duì)應(yīng)的查詢計(jì)劃執(zhí)行器（PlanExecutor），構(gòu)建PlanExecutor時(shí)候，用了一個(gè)特殊的執(zhí)行Stage, 即ProxyStage完成對(duì)整個(gè)Pipeline的封裝，此外也會(huì)把對(duì)應(yīng)的游標(biāo)ID返回給Mongos節(jié)點(diǎn)。

客戶端利用從Mongos節(jié)點(diǎn)拿到游標(biāo)ID, 在該游標(biāo)上不斷的執(zhí)行g(shù)etMore請(qǐng)求，服務(wù)端收到getMore請(qǐng)求后，最后通過(guò)cursor的next調(diào)用，轉(zhuǎn)發(fā)請(qǐng)求到shard節(jié)點(diǎn)，拿到數(shù)據(jù)后，歸并后返回可客戶端，完成了整個(gè)Change Streams事件的訂閱。

Shard上pipeline具體執(zhí)行的細(xì)節(jié)不在本文重點(diǎn)介紹范圍，這些就不詳細(xì)展開(kāi)了。

原生Change Stream目前使用上有如下限制：

1. 支持DDL事件不完善

Change Stream目前支持的事件如下：

• Insert Event
• Update Event
• Replace Event
• Delete Event
• Drop Event
• Rename Event
• DropDatabase Event
• invalidate Event

顯然上述事件并沒(méi)有完全覆蓋MongoDB內(nèi)部全部的數(shù)據(jù)變更的事件。

此外，對(duì)于在集合上監(jiān)聽(tīng)的Change Streams， 當(dāng)出現(xiàn)集合或者所屬的DB被刪除后，會(huì)觸發(fā)一個(gè)invalidate Event， 該事件會(huì)把Change Streams的cursor關(guān)閉掉，導(dǎo)致Change Streams無(wú)法繼續(xù)進(jìn)行，對(duì)于通過(guò)Change Streams來(lái)實(shí)現(xiàn)容災(zāi)的場(chǎng)景，顯然是不夠友好的，需要重新建立新的Change Streams監(jiān)聽(tīng)。

2. 事件拉取性能有待提升

如上述分析，當(dāng)前的Change Streams請(qǐng)求發(fā)到Mongos節(jié)點(diǎn)后，通過(guò)單線程的方式向每個(gè)Shard節(jié)點(diǎn)發(fā)送異步請(qǐng)求命令來(lái)完成數(shù)據(jù)的拉取，并做數(shù)據(jù)歸并，如果將該方式替換為多線程并發(fā)拉取，對(duì)于分片表來(lái)說(shuō)，性能會(huì)有提升。

三、 并行Change Streams架構(gòu)和原理

3.1 并發(fā)Change Streams架構(gòu)介紹 

針對(duì)上述的一些使用限制，我們結(jié)合實(shí)際客戶使用需求，提出一種新的并發(fā)Change Streams（Parallel Change Streams）的方式，來(lái)嘗試解決上述問(wèn)題。

為了提升原生Change Streams的性能，我們?cè)贛ongos 節(jié)點(diǎn)引入如下幾個(gè)新的組件：

• Change Streams Buffer

與Shard是一對(duì)一的關(guān)系。每個(gè)Change Streams Buffer 默認(rèn)1GB，在Buffer滿之前，該Buffer無(wú)條件的向?qū)?yīng)的Shard(secondary節(jié)點(diǎn))拉取Change Streams數(shù)據(jù)。

• Merged Queue

Merged Queue是一個(gè)內(nèi)存隊(duì)列，是Change Streams Buffer的消費(fèi)者，是 Bucket的生產(chǎn)者。Merged Queue 歸并所有Shard的Change Streams Buffer，并等待合適的時(shí)機(jī)按照規(guī)則放入對(duì)應(yīng)Client的Bucket。

• Bucket

Bucket 是一個(gè)內(nèi)存隊(duì)列，是MergedQueue的消費(fèi)者，是Client的生產(chǎn)者。每個(gè)Client對(duì)應(yīng)一個(gè)Bucket。每個(gè)Bucket維護(hù)該Bucket內(nèi)所有文檔的的集合。

• Merged Queue 與Bucket的交互過(guò)程

Merged Queue不停的從頭部拿出盡可能多的數(shù)據(jù)，并從前往后的按照hash(document.ns)%n的規(guī)則放入對(duì)應(yīng)的Bucket, document.ns是指這個(gè)文檔的NameSpace， 所以同一個(gè)集合的數(shù)據(jù)一定在一個(gè)Bucket里面。

3.2 對(duì)DDL事件的增強(qiáng) 

并發(fā)Change Stream除了支持原生的Change Stream外，還新增支持如下事件：

• CreateCollection Event
• CollMod Event
• CreateIndex Event
• Drop Index Event
• CreateView Event
• DropView Event
• ShardCollection Event

本文以ShardCollection為例來(lái)說(shuō)明如何實(shí)現(xiàn)新增DDL事件的支持：

當(dāng)執(zhí)行ShardCollection命令的時(shí)候，Config節(jié)點(diǎn)會(huì)向該集合的主Shard發(fā)送一個(gè)shardsvrShardCollection命令，主Shard收到改請(qǐng)求后，我們?cè)谠撁畹奶幚砹鞒讨杏涗浟艘粋€(gè)type為noop的oplog, 并把該命令的詳細(xì)內(nèi)容寫入到oplog的o2字段里面，以此來(lái)實(shí)現(xiàn)shardcollecton事件的追蹤。

之后在處理Change Streams流程的pipeline中，我們對(duì)noop事件進(jìn)行分析，如果其中內(nèi)容包括了shardCollection事件相關(guān)的標(biāo)記，則提取該事件，并返回給上層。

3.3 如何使用 

1 如果想創(chuàng)建并發(fā)change Stream，需要先通過(guò)如下命令創(chuàng)建bucket和cursor：

db.runCommand( 
 
{ 
 
    parallelChangeStream: 1,  
 
    nBuckets: Required,<int>,  
 
    nsRegex: Optional,<Regex>,  
 
startAtOperationTime: Optional,<Timestamp>,  
 
}) 

參數(shù)說(shuō)明如下：

parallelChangeStream ：開(kāi)啟并行changeStream

nBuckets：要?jiǎng)?chuàng)建的bucket的數(shù)目

nsRegex：可選，定義要訂閱的集合，一個(gè)正則表達(dá)式。

startAtOperationTime：可選，表示訂閱的事件從哪個(gè)時(shí)間點(diǎn)開(kāi)始。

返回值：

"cursors" : [ 
 
    NumberLong("2286048776922859088"), 
 
    NumberLong("2286048779108179584"), 
 
    NumberLong("2286048780088774662"), 
 
    NumberLong("2286048777169702425"), 
 
    NumberLong("2286048779233363970"), 
 
    NumberLong("2286048779250024945"), 
 
    NumberLong("2286048776628281242"), 
 
    NumberLong("2286048778209018113"), 
 
    NumberLong("2286048778833886224"), 
 
    NumberLong("2286048777951363227") 
 
] 

Cursors ：返回的Mongos側(cè)的Cursor ID。

當(dāng)獲取到所有Cursor ID后，客戶端就可以并發(fā)的(每個(gè)CursorId一個(gè)線程)通過(guò)getMore命令不斷的從服務(wù)端拉取結(jié)果了。

斷點(diǎn)續(xù)傳

ParallelChangeStream的斷點(diǎn)續(xù)傳通過(guò)startAtOperationTime實(shí)現(xiàn)，由于每個(gè)cursor的消費(fèi)進(jìn)度不一樣，恢復(fù)的斷點(diǎn)應(yīng)該選用n個(gè)cursor的消費(fèi)值的最小值。

四、性能對(duì)比

針對(duì)新的Parallel Change Stream和原生的Change Streams ，我們做了較長(zhǎng)時(shí)間的對(duì)比測(cè)試分析，所有測(cè)試場(chǎng)景采用的測(cè)試實(shí)例如下：

實(shí)例規(guī)格：4U16G， 2個(gè)Shard（副本集） ，2個(gè)Mongos，

磁盤容量：500G

測(cè)試數(shù)據(jù)模型：通過(guò)YCSB 預(yù)置數(shù)據(jù)，單條記錄1K , 單個(gè)分片表1000w條記錄。

下面分幾個(gè)場(chǎng)景分別介紹：

1. 集群模式1分片表場(chǎng)景測(cè)試 

測(cè)試方法：

1） 創(chuàng)建一個(gè)Hash分片的集合，預(yù)置16 Chunk

2） 啟動(dòng)YCSB , 對(duì)該集合進(jìn)行Load數(shù)據(jù)操作，Load數(shù)據(jù)量為1000w ，設(shè)置的Oplog足夠大，保證這些操作還在Oplog中

3） 分別啟動(dòng)原生Change Streams 和 Parallel Change Streams，通過(guò)指定startAtOperationTime來(lái)觀察訂閱1000w條記錄分別需要花費(fèi)的時(shí)間。

4） 由于是單個(gè)表, nBuckets 為1

測(cè)試數(shù)據(jù)如下：

	讀取總數(shù)據(jù)量	花費(fèi)總時(shí)間（ms ）	TPS( 個(gè)/s)
Change  Streams	1000w	432501	23148
Parallel Change Streams(1 bucket)	1000w	184437	54361

2. 集群模2分片表場(chǎng)景測(cè)試 

測(cè)試方法：

1） 創(chuàng)建2個(gè)Hash分片的集合，預(yù)置16 Chunk

2） 啟動(dòng)YCSB , 同時(shí)對(duì)這2個(gè)集合進(jìn)行Load數(shù)據(jù)操作，每個(gè)集合Load數(shù)據(jù)量為1000w ，設(shè)置的Oplog足夠大，保證這些操作還在Oplog中

3） 分別啟動(dòng)原生Change Streams和Parallel Change Streams，通過(guò)指定startAtOperationTime來(lái)觀察訂閱4000w條記錄分別需要花費(fèi)的時(shí)間。

4） 由于是2個(gè)表, nBuckets 為2

測(cè)試數(shù)據(jù)如下：

	讀取總數(shù)據(jù)量	花費(fèi)總時(shí)間（ms ）	TPS( 個(gè)/s)
Change  Streams	4000w	2151792	18484
Parallel  Change Streams	4000w	690776	55248

3. 集群模式4分片表場(chǎng)景測(cè)試 

測(cè)試方法：

1） 創(chuàng)建4個(gè)Hash分片的集合，預(yù)置16 Chunk

2） 啟動(dòng)YCSB , 同時(shí)對(duì)這4個(gè)集合進(jìn)行Load數(shù)據(jù)操作，每個(gè)集合Load數(shù)據(jù)量為1000w ，設(shè)置的Oplog足夠大，保證這些操作還在Oplog中

3） 分別啟動(dòng)原生Change Streams和Parallel Change Streams，通過(guò)指定startAtOperationTime來(lái)觀察訂閱4000w條記錄分別需要花費(fèi)的時(shí)間。

4） 由于是4個(gè)表, nBuckets 為4

測(cè)試數(shù)據(jù)如下：

	讀取總數(shù)據(jù)量	花費(fèi)總時(shí)間（ms ）	TPS( 個(gè)/s)
Change  Streams	4000w	2151792	18596
Parallel  Change Streams	4000w	690776	56577

總結(jié)：通過(guò)實(shí)際測(cè)試可以看出來(lái)， Parallel Change Streams這種方式性能有極大的提升，實(shí)際上我們后續(xù)根據(jù)實(shí)例規(guī)格，通過(guò)調(diào)整內(nèi)部Bucket和Buffer的緩存大小，性能還可以繼續(xù)提升，同時(shí)隨著分片表數(shù)據(jù)量和Shard節(jié)點(diǎn)數(shù)量的變多，和原生Change Streams 的性能優(yōu)勢(shì)會(huì)更加明顯。

五、并發(fā)Change Streams使用場(chǎng)景分析

并發(fā)Change Streams非常適合在MongoDB集群的容災(zāi)場(chǎng)景，應(yīng)用可以有針對(duì)性的設(shè)置對(duì)特定的集合或者DB進(jìn)行監(jiān)聽(tīng)，可以實(shí)時(shí)的感知到源端實(shí)例的數(shù)據(jù)變化，并快速的應(yīng)用到目標(biāo)端，整體實(shí)現(xiàn)較低RPO。

此外，并發(fā)Change Streams也可以應(yīng)用到PITR場(chǎng)景中， 通過(guò)并發(fā)Change Streams良好的性能，實(shí)時(shí)實(shí)現(xiàn)動(dòng)態(tài)數(shù)據(jù)的跟蹤并記錄，使得PITR的可恢復(fù)時(shí)間更短。

六、未來(lái)展望

當(dāng)前的并行Change Streams的實(shí)現(xiàn)中，merge queue中的事件分發(fā)到bucket的事件中，我們采用的策略是基于事件的NameSpace的HASH值，傳遞給對(duì)應(yīng)的bucket中，這種策略對(duì)于單集合的場(chǎng)景，性能優(yōu)化有限，后續(xù)我們計(jì)劃同時(shí)提供基于事件的ID內(nèi)容的HASH值，把事件分發(fā)到不同的bucket中，這種方式能進(jìn)一步的提升系統(tǒng)并發(fā)性能，帶來(lái)更好的性能優(yōu)化效果。

七、總結(jié)

通過(guò)引入一種新的并發(fā)Change Streams的方式，支持更多類別的MongoDB事件的訂閱，同時(shí)在事件監(jiān)聽(tīng)的性能方面相比原生有較大的提高，可以廣泛應(yīng)用在數(shù)據(jù)庫(kù)實(shí)例容災(zāi)， PITR，數(shù)據(jù)在線遷移業(yè)務(wù)場(chǎng)景中，為客戶帶來(lái)更好的體驗(yàn)。