ApsaraDB for MongoDB：使用MongoDB變更流（Change Stream）即時捕獲資料變更

監聽更多的DDL事件

前提條件

MongoDB 6.0及以上版本（升級指南）。

操作步驟

1. 使用 mongosh 串連資料庫。

2. 執行命令watch，設定 showExpandedEvents: true：

   // mongo shell or mongosh v1.x
   cursor = db.getSiblingDB("test").watch([],
     {
       showExpandedEvents: true       // 啟用更多DDL 事件監聽
     }
   );
   cursor.next();

結果驗證

1. 在新的SQL視窗執行變更SQL，如db.createCollection("myCollection1")。

2. 觀察原SQL視窗輸出上述執行SQL相關資訊。

   mg xxx test> cursor = db.getSiblingDB("test").watch([],
   ... {
   ...     showExpandedEvents: true     // 啟用 更多 DDL 事件監聽
   ... }
   ... );
   ... cursor.next();
   Warning: If there are no documents in the batch, next will block. Use tryNext if you want to check if there are any documents without waiting.
   {
     _id: {
       _data: '8268777614000000012B042C0100296E5A100434919C64814940F3A61F9D97A3F60D5C463C6Fxxx03C63726561746500466F70657274696F6E4465737372697074696F6E0046466964496E6465787800461E76002B04466B657990046E5F6964002B02003C6E616D65003C5F69645F000000004'
     },
     operationType: 'create',
     clusterTime: Timestamp({ t: 1752659476, i: 1 }),
     collectionUUID: UUID('34919c64-8149-40f3-a61f-9xxx'),
     wallTime: ISODate('2025-07-16T09:51:16.043Z'),
     ns: { db: 'test', coll: 'myCollection1' },
     operationDescription: { idIndex: { v: 2, key: { _id: 1 }, name: '_id_' } }
   }

開啟前鏡像（Pre-image）

MongoDB的pre-image是指文檔在修改或刪除前的完整狀態快照，用於記錄資料變更前的原始值。

前提條件

MongoDB 6.0及以上版本（升級指南）。

操作步驟

1. 開啟資料庫維度前鏡像 ：

   db.adminCommand({
     setClusterParameter: {
       changeStreamOptions: {
         preAndPostImages: { expireAfterSeconds: "off" } // "off" 表示使用oplog保留時間
       }
     }
   })

2. 開啟集合維度前鏡像 ：

   說明

   若需為某資料庫內所有集合啟用前鏡像，則必須先開啟資料庫維度前鏡像，然後在該庫的每個集合上單獨啟用集合級前鏡像功能。

   修改現有集合

   db.runCommand({
     collMod: "myCollection",
     changeStreamPreAndPostImages: { enabled: true }
   })

   建立新集合時指定

   db.createCollection("myCollection", { changeStreamPreAndPostImages: { enabled: true }})

3. 建立變更流監聽 （指定Pre-image選項）：

   // 在目的地組合上建立監聽
   cursor = db.getCollection("myCollection").watch([],
     {
       fullDocument: 'required', // 或 'whenAvailable'
       fullDocumentBeforeChange: 'required' // 或 'whenAvailable'
     }
   )
   cursor.next();

   • required: 服務端必須返回Pre/Post-image，否則報錯。

   • whenAvailable: 服務端儘力返回，不保證一定有。

結果驗證

1. 檢查資料庫維度開關是否已開啟：

   db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

   開啟成功後輸出資訊。

   test> db.adminCommand( { getClusterParameter: "changeStreamOptions" } )
   {
     clusterParameters: [
       {
         _id: 'changeStreamOptions',
         clusterParameterTime: Timestamp({ t: 1752655937, i: 1 }),
         preAndPostImages: { expireAfterSeconds: Long('100') }
       }
     ],
     ok: 1,
     '$clusterTime': {
       clusterTime: Timestamp({ t: 1752656717, i: 1 }),
       signature: {
         hash: Binary.createFromBase64('xxx=', 0),
         keyId: Long('xxx')
       }
     },
     operationTime: Timestamp({ t: 1752656717, i: 1 })
   }

2. 檢查集合配置：

   db.getCollectionInfos({name: "myCollection"}) // 或 db.runCommand({listCollections: 1})

   預期輸出：在返回的文檔中找到類似 "options" : { "changeStreamPreAndPostImages" : { "enabled" : true } } 的欄位。 

   test> db.getCollectionInfos({name: "myCollection"})
   [
     {
       name: 'myCollection',
       type: 'collection',
       options: { changeStreamPreAndPostImages: { enabled: true } },
       info: {
         readOnly: false,
         uuid: UUID('55ed1b7c-7575-4ba3-8afa-xxx')
       },
       idIndex: { v: 2, key: { _id: 1 }, name: '_id_' }
     }
   ]

3. 在另一個mongosh視窗更新 myCollection 中的文檔。

4. 觀察 cursor 返回的事件，應包含 fullDocumentBeforeChange 欄位（變更前文檔）。

   ... cursor.next();
   {
     _id: {
       _data: '8268786B15000000012B042C0100296E5A100455ED1B7C75754BA38AFA0C5B56A360BC463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B65790046xxx697CDDF6EE279xxx'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1752722197, i: 1 }),
     wallTime: ISODate('2025-07-17T03:16:37.626Z'),
     fullDocument: {
       _id: ObjectId('6878697cddfxxx'),
       name: 'test',
       count: 111
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('6878697cddfxxx') },
     updateDescription: {
       updatedFields: { count: 111 },
       removedFields: [],
       truncatedArrays: []
     },
     fullDocumentBeforeChange: { _id: ObjectId('6878697cddf6xxx'), name: 'test' }
   }

更多資訊參考Change Streams with Document Pre- and Post-Images。

開啟後鏡像（Post-image）

MongoDB的Post-image是變更發生後文檔的最新完整狀態快照，用於記錄變更後的完整文檔內容。

前提條件

MongoDB 3.6及以上版本（升級指南）。

操作步驟

執行watch命令時設定fullDocument: 'updateLookup'。

cursor = db.getSiblingDB("test").myCollection.watch([], 
  {
    fullDocument: 'updateLookup'
  }
);
cursor.next();
              

結果驗證

1. 在另一個mongosh視窗新增或更新 myCollection 中的文檔。

2. 觀察 cursor 返回的事件，應包含 fullDocument 欄位（變更後文檔）。

   xxx [primary] test> cursor = db.getSiblingDB("test").myCollection.watch([],
   ... {
   ...     fullDocument: 'updateLookup'
   ... }
   ... );
   ... cursor.next();
   [...
   {
     _id: {
       _data: 'xxx100296E5A10040D6C3FBC28484F08240555576066945463C6F7065726174696F6E54797065003C757064617465500046646F63756D656E744B6579005F6964004B657390046645F696400646881ADE741E7D4B638ED7CA1000004'
     },
     operationType: 'update',
     clusterTime: Timestamp({ t: 1753329166, i: 1 }),
     wallTime: ISODate('2025-07-24T03:52:46.966Z'),
     fullDocument: {
       _id: ObjectId('xxx8ed7ca1'),
       name: 'test1',
       age: 12,
       count: 2222
     },
     ns: { db: 'test', coll: 'myCollection' },
     documentKey: { _id: ObjectId('xxx7ca1') },
     updateDescription: {
       updatedFields: { count: 2222 },
       removedFields: [],
       truncatedArrays: []
     }
   }]

處理超大變更事件 (>16MB)

前提條件

MongoDB 7.0及以上版本（升級指南）。

操作步驟 

在 watch() 的管道中包含 $changeStreamSplitLargeEvent 階段：

myChangeStreamCursor = db.myCollection.watch(
  [ { $changeStreamSplitLargeEvent: {} } ], // 添加切分階段
  {
    fullDocument: "required",
    fullDocumentBeforeChange: "required"
  }
)

結果驗證

• 執行一個會導致變更事件 >16MB 的操作（如更新一個包含超大數組的文檔）。

• 觀察返回的事件流會被拆分成多個連續的fragment事件，最終以一個fragment事件結束。

降低前鏡像儲存開銷

預設Pre-image隨oplog到期。可設定更短的到期時間以節省空間的：

db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 100 } // 單位：秒
    }
  }
})

1. 為什麼 Change Stream 的慢日誌總是顯示 COLLSCAN？

這是正常且符合預期的行為，無需最佳化。Change Stream的cursor最終會建立在local.oplog.rs上（只有這一個地方記錄了執行個體所有的修改），而該表沒有索引。所以固定是COLLSCAN，無法避免也沒有最佳化空間。

如果您在尋找待最佳化查詢語句時使用了COLLSCAN作為日誌過濾的關鍵字，推薦也額外過濾掉$changeStream關鍵字。

推薦在遇到changeStream效能問題（比如下遊消費拉取存在延遲上漲）時才需要關注Change Stream相關的慢日誌。

2. 為什麼分區執行個體上對於非分區表建立監聽，也能在除了primary shard之外的其他分區上觀察到changeStream cursor？

為了應對可能存在的分區（shardCollection）操作。您可以隨時對一個已經建立監聽的非分區表執行shardCollection操作來將其資料分布到所有分區上（movePrimary類似）。提前在其他分區上建立Change Stream cursors就是為了應對這種情況。通常這些分區上的cursor並不會返回任何變更事件，效能開銷也比較小。

與之類似的，為了應對可能存在的addShard/removeShard操作，分區執行個體上建立監聽時，mongos也會到Config Server上去建立相關的cursor。

3. 為什麼建立Change Stream cursor時指定了readPreference:secondary，但這個cursor的慢日誌卻出現在primary中？

遊標（cursor）建立後會“固定”在某個節點上，不會因節點角色變化而自動遷移。因此一旦Change Stream的cursor建立之後，就固定在某個mongod節點（主或從節點）上並持續通過getMore消費。當發生過主從切換、變更配置、遷移或其他可能導致主從切換的事件時，原本在secondary上的cursor可能會轉移到primary上。

如果不期望這部分負載持續在primary節點上消耗資源，則可以通過killCursors的方式來清理相關的changeStream cursor，下遊消費邏輯處理會根據resumeToken並結合readPreference將cursor恢複到secondary節點上去，此操作不會導致變更事件的丟失或中斷。

db.runCommand(
   {
     killCursors: <collection>,
     cursors: [ <cursor id1>, ... ], comment: <any>
   }
)
db.getSiblingDB("<testDB>").runCommand( { killCursors: "<testColl>", cursors: [NumberLong("2452840976689696187") ] } ) 

4. 為什麼在mongos上看到大量關於Change Stream的長達1000ms的慢日誌？

2020-08-26T04:34:45.045+0000 I COMMAND [conn21283] command altconfig-b2b-perf.oplog command: getMore \{ getMore: 3513599116181216748, collection: "oplog", $db: "altconfig-b2b-perf", $clusterTime: { clusterTime: Timestamp(1598416483, 1), signature: { hash: BinData(0, EC3841EB1FB7A34F897688BB5983E32E2ADF6763), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } originatingCommand: \{ find: "oplog", filter: { timestamp: { $gte: new Date(1597895390009) } }, tailable: true, awaitData: true, $db: "altconfig-b2b-perf", $clusterTime: \{ clusterTime: Timestamp(1597895487, 1), signature: { hash: BinData(0, AAAE33D5688F935C70469CBDB8EB1F6882749A40), keyId: 6855385090000683010 } }, lsid: \{ id: UUID("72864d11-c0f8-48bb-823e-de5f18a9c409") } } nShards:1 cursorid:3513599116181216748 numYields:0 nreturned:0 reslen:237 protocol:op_msg 1000ms

這是舊版本 MongoDB （6.0 之前） 的正常現象，它代表的是等待逾時而非效能瓶頸。在分區執行個體架構中，mongos到分區建立的Change Stream cursors會指定tailable:true、awaitData:true以及maxTimeMS:1000來儘可能在1000ms內返回更多的變更事件。

在6.0之前的大版本中，mongos上會記錄這些長達1000ms的慢日誌，存在誤導客戶的可能性。官方已經對相關行為進行了最佳化，細節可參考SERVER-50559。

5. 開啟前鏡像（Pre-image）後資料存放區在哪裡？

前鏡像儲存在config.system.preimages這個系統集合中，存在於每個 mongod 資料節點上（不是 Config Server）：

• 複本集：存在於各資料節點（primary + secondary 都有，隨 oplog 複製）。

• 分區叢集：存在於每個 shard 的 mongod 上，各 shard 隔離儲存區 (Isolated Storage)自己處理的文檔的前鏡像。

後鏡像不額外儲存，會從 Change Stream event 中即時構造返回。

6. 開啟前鏡像（Pre-image）後會有什麼開銷？

• 儲存開銷：config.system.preimages集合的大小會持續增長，如果不配置到期策略或 oplog 視窗很大，可能佔用大量磁碟空間。

• 寫入開銷：每次update/replace/delete操作都會額外寫一份完整的舊文檔到config.system.preimages，相當於寫 I/O 翻倍（對大文檔尤其明顯）。

• WT cache 壓力：config.system.preimages集合的資料也會進入 WiredTiger cache，擠占正常業務資料的緩衝空間。

• oplog 記錄：因為需要複製到複本集的從節點，因此前鏡像文檔的寫入也會產生 oplog 條目。

• 16MB 限制：文檔大小比較大時，包含前鏡像的變更事件可能會超過 16MB 限制而導致變更流報錯。6.0 版本由於還不支援$changeStreamSplitLargeEvent階段，資料庫無法自行處理，需要業務側介入。