MongoDB Change Streams性能優化實踐

作者：佚名 2021-09-24 14:02:53

本文介紹一種新的事件訂閱方式，來完善上述不足，并探討通過并發預讀的方式，來提升原生Change Streams的性能。

簡介：基于MongoDB的應用程序通過Change Streams功能可以方便的實現對某個集合，數據庫或者整個集群的數據變更的訂閱，極大的方便了應用對數據庫變化的感知，但是當前Change Streams對部分數據的變化并沒有提供對應的事件（創建索引，刪除索引，shardCollection）等，本文介紹一種新的事件訂閱方式，來完善上述不足，并探討通過并發預讀的方式，來提升原生Change Streams的性能。

一、前言

MongoDB作為一款優秀的NOSQL數據庫，支持海量存儲，查詢能力豐富以及優秀的性能和可靠性，當前大部分云廠商都提供了兼容MongoDB協議的服務，用戶使用廣泛，深受國內外用戶和企業的認可。

MongoDB從3.6版本開始提供了Change Stream特性，通過該特性，應用程序可以實時的訂閱特定集合、庫、或整個集群的數據變更事件，相比該特性推出之前通過監聽oplog的變化來實現對數據變更的感知，非常的易用，該特性同時支持副本集和集群場景。

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

本文通過深入分析當前的Change Stream實現機制，結合客戶實際使用場景，提出了一種新的多并發預讀的事件監聽方式，來解決上述問題，并應用到客戶實際遷移和數據庫容災的場景中。

二、Change Steams 機制介紹

Change Streams支持對單個集合，DB，集群進行事件訂閱，當業務程序通過watch的方式發起訂閱后，背后發生了什么，讓我們一起來分析一下。

Change Streams內部實現是通過Aggregate的方式實現的，所以watch背后，對應的是客戶端向MongoDB Server發起了一個Aggregate命令，且對Aggregate的pipeline 參數中，添加了一個$changeStream的Stage, 結合客戶端其他參數，一起發給MongoDB Server。

當Mongo Server收到Aggregate命令后，解析后，會根據具體的請求，組合一個新的Aggregate命令，并把該命令發給對應的Shard節點，同時會在游標管理器（CursorManger）中注冊一個新的游標（cursor），并把游標Id返回給客戶端。

當Shard Server端收到Aggregate命令后，構建pipeline流水線，并根據pipeline參數中包括了Change Steams參數，確定原始掃描的集合為oplog，并創建對該集合上掃描數據的原始cursor, 和對應的查詢計劃執行器（PlanExecutor），構建PlanExecutor時候，用了一個特殊的執行Stage, 即ProxyStage完成對整個Pipeline的封裝，此外也會把對應的游標ID返回給Mongos節點。

客戶端利用從Mongos節點拿到游標ID, 在該游標上不斷的執行getMore請求，服務端收到getMore請求后，最后通過cursor的next調用，轉發請求到shard節點，拿到數據后，歸并后返回可客戶端，完成了整個Change Streams事件的訂閱。

Shard上pipeline具體執行的細節不在本文重點介紹范圍，這些就不詳細展開了。

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

1. 支持DDL事件不完善

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

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

顯然上述事件并沒有完全覆蓋MongoDB內部全部的數據變更的事件。

此外，對于在集合上監聽的Change Streams，當出現集合或者所屬的DB被刪除后，會觸發一個invalidate Event，該事件會把Change Streams的cursor關閉掉，導致Change Streams無法繼續進行，對于通過Change Streams來實現容災的場景，顯然是不夠友好的，需要重新建立新的Change Streams監聽。

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

如上述分析，當前的Change Streams請求發到Mongos節點后，通過單線程的方式向每個Shard節點發送異步請求命令來完成數據的拉取，并做數據歸并，如果將該方式替換為多線程并發拉取，對于分片表來說，性能會有提升。

三、并行Change Streams架構和原理

3.1 并發Change Streams架構介紹

針對上述的一些使用限制，我們結合實際客戶使用需求，提出一種新的并發Change Streams（Parallel Change Streams）的方式，來嘗試解決上述問題。

為了提升原生Change Streams的性能，我們在Mongos 節點引入如下幾個新的組件：

Change Streams Buffer

與Shard是一對一的關系。每個Change Streams Buffer 默認1GB，在Buffer滿之前，該Buffer無條件的向對應的Shard(secondary節點)拉取Change Streams數據。

Merged Queue

Merged Queue是一個內存隊列，是Change Streams Buffer的消費者，是 Bucket的生產者。Merged Queue 歸并所有Shard的Change Streams Buffer，并等待合適的時機按照規則放入對應Client的Bucket。

Bucket

Bucket 是一個內存隊列，是MergedQueue的消費者，是Client的生產者。每個Client對應一個Bucket。每個Bucket維護該Bucket內所有文檔的的集合。

Merged Queue 與Bucket的交互過程

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

3.2 對DDL事件的增強

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

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

本文以ShardCollection為例來說明如何實現新增DDL事件的支持：

當執行ShardCollection命令的時候，Config節點會向該集合的主Shard發送一個shardsvrShardCollection命令，主Shard收到改請求后，我們在該命令的處理流程中記錄了一個type為noop的oplog, 并把該命令的詳細內容寫入到oplog的o2字段里面，以此來實現shardcollecton事件的追蹤。

之后在處理Change Streams流程的pipeline中，我們對noop事件進行分析，如果其中內容包括了shardCollection事件相關的標記，則提取該事件，并返回給上層。

3.3 如何使用

1 如果想創建并發change Stream，需要先通過如下命令創建bucket和cursor：

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

參數說明如下：

parallelChangeStream ：開啟并行changeStream

nBuckets：要創建的bucket的數目

nsRegex：可選，定義要訂閱的集合，一個正則表達式。

startAtOperationTime：可選，表示訂閱的事件從哪個時間點開始。

返回值：

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

Cursors ：返回的Mongos側的Cursor ID。

當獲取到所有Cursor ID后，客戶端就可以并發的(每個CursorId一個線程)通過getMore命令不斷的從服務端拉取結果了。

斷點續傳

ParallelChangeStream的斷點續傳通過startAtOperationTime實現，由于每個cursor的消費進度不一樣，恢復的斷點應該選用n個cursor的消費值的最小值。

四、性能對比

針對新的Parallel Change Stream和原生的Change Streams ，我們做了較長時間的對比測試分析，所有測試場景采用的測試實例如下：

實例規格：4U16G， 2個Shard（副本集），2個Mongos，

磁盤容量：500G

測試數據模型：通過YCSB 預置數據，單條記錄1K , 單個分片表1000w條記錄。

下面分幾個場景分別介紹：

1. 集群模式1分片表場景測試

測試方法：

1）創建一個Hash分片的集合，預置16 Chunk

2）啟動YCSB , 對該集合進行Load數據操作，Load數據量為1000w ，設置的Oplog足夠大，保證這些操作還在Oplog中

3）分別啟動原生Change Streams 和 Parallel Change Streams，通過指定startAtOperationTime來觀察訂閱1000w條記錄分別需要花費的時間。

4）由于是單個表, nBuckets 為1

測試數據如下：

	讀取總數據量	花費總時間（ms ）	TPS( 個/s)
Change Streams	1000w	432501	23148
Parallel Change Streams(1 bucket)	1000w	184437	54361

2. 集群模2分片表場景測試

測試方法：

1）創建2個Hash分片的集合，預置16 Chunk

2）啟動YCSB , 同時對這2個集合進行Load數據操作，每個集合Load數據量為1000w ，設置的Oplog足夠大，保證這些操作還在Oplog中

3）分別啟動原生Change Streams和Parallel Change Streams，通過指定startAtOperationTime來觀察訂閱4000w條記錄分別需要花費的時間。

4）由于是2個表, nBuckets 為2

測試數據如下：

	讀取總數據量	花費總時間（ms ）	TPS( 個/s)
Change Streams	4000w	2151792	18484
Parallel Change Streams	4000w	690776	55248

3. 集群模式4分片表場景測試

測試方法：

1）創建4個Hash分片的集合，預置16 Chunk

2）啟動YCSB , 同時對這4個集合進行Load數據操作，每個集合Load數據量為1000w ，設置的Oplog足夠大，保證這些操作還在Oplog中

3）分別啟動原生Change Streams和Parallel Change Streams，通過指定startAtOperationTime來觀察訂閱4000w條記錄分別需要花費的時間。

4）由于是4個表, nBuckets 為4

測試數據如下：

	讀取總數據量	花費總時間（ms ）	TPS( 個/s)
Change Streams	4000w	2151792	18596
Parallel Change Streams	4000w	690776	56577

總結：通過實際測試可以看出來， Parallel Change Streams這種方式性能有極大的提升，實際上我們后續根據實例規格，通過調整內部Bucket和Buffer的緩存大小，性能還可以繼續提升，同時隨著分片表數據量和Shard節點數量的變多，和原生Change Streams 的性能優勢會更加明顯。

五、并發Change Streams使用場景分析

并發Change Streams非常適合在MongoDB集群的容災場景，應用可以有針對性的設置對特定的集合或者DB進行監聽，可以實時的感知到源端實例的數據變化，并快速的應用到目標端，整體實現較低RPO。

此外，并發Change Streams也可以應用到PITR場景中，通過并發Change Streams良好的性能，實時實現動態數據的跟蹤并記錄，使得PITR的可恢復時間更短。

六、未來展望

當前的并行Change Streams的實現中，merge queue中的事件分發到bucket的事件中，我們采用的策略是基于事件的NameSpace的HASH值，傳遞給對應的bucket中，這種策略對于單集合的場景，性能優化有限，后續我們計劃同時提供基于事件的ID內容的HASH值，把事件分發到不同的bucket中，這種方式能進一步的提升系統并發性能，帶來更好的性能優化效果。

七、總結

通過引入一種新的并發Change Streams的方式，支持更多類別的MongoDB事件的訂閱，同時在事件監聽的性能方面相比原生有較大的提高，可以廣泛應用在數據庫實例容災， PITR，數據在線遷移業務場景中，為客戶帶來更好的體驗。

責任編輯：張燕妮來源： MongoDB中文社區

性能優化實踐

MongoDB Change Streams性能優化實踐

一、前言

二、Change Steams 機制介紹

1. 支持DDL事件不完善

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

三、 并行Change Streams架構和原理

3.1 并發Change Streams架構介紹

3.2 對DDL事件的增強

3.3 如何使用

參數說明如下：

返回值：

斷點續傳

四、性能對比

1. 集群模式1分片表場景測試

測試方法：

測試數據如下：

2. 集群模2分片表場景測試

測試方法：

測試數據如下：

3. 集群模式4分片表場景測試

測試方法：

測試數據如下：

五、并發Change Streams使用場景分析

六、未來展望

七、總結

三、并行Change Streams架構和原理