MongoDB 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，數據在線遷移業務場景中，為客戶帶來更好的體驗。
