[[425468]]

簡介：基于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目前的性能問題，官方也有過探討http://www.ythuaji.com.cn/uploads/allimg/pnlfhyqtzfk

本文通過深入分析當前的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

測試數據如下：

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

測試數據如下：

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

測試數據如下：

總結：通過實際測試可以看出來， 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，數據在線遷移業務場景中，為客戶帶來更好的體驗。