MongoDB Change Stream更新操作获取变更前旧值的可行方案咨询

方案1：升级到MongoDB 6.0+（最优原生方案） #

MongoDB 6.0及以上版本已经原生支持变更前旧值获取，只需要在开启Change Stream时添加fullDocumentBeforeChange: "required"参数即可，所有更新、删除事件都会自动返回变更前的完整文档，无需额外开发逻辑，完全规避你提到的缓存冷启动、多watcher同步问题。
参数使用示例：

const changeStream = collection.watch([], {
  fullDocument: 'updateLookup',
  fullDocumentBeforeChange: 'required'
})

每个变更事件的fullDocumentBeforeChange字段就是变更前的旧值。

方案2：低版本MongoDB兼容方案 #

如果无法升级到6.0以上版本，可以选择两种实现路径：

路径A：优化缓存方案（复杂度低） #

针对你提到的两个缓存缺陷，可以通过以下方式修复：

• 解决首次变更无旧值问题：watcher启动监听前，先全量扫描一次监听范围的所有文档，将所有文档的当前值写入缓存后再启动Change Stream监听，从源头避免冷启动无缓存的情况。如果集合数据量过大，可以配合Change Stream的startAtOperationTime参数，将缓存快照的时间点和Change Stream启动的时间点对齐，避免快照写入期间发生的变更丢失。
• 解决多watcher缓存同步问题：缓存更新操作改成基于Change Stream事件的clusterTime（集群时间戳）做版本校验，只有当前事件的时间戳大于缓存中存储的该文档对应时间戳时，才允许读取旧缓存值并更新缓存，避免低版本的事件覆盖高版本的缓存数据。示例逻辑：

  const event = getChangeStreamEvent()
  const cacheKey = `doc:${event.documentKey._id}`
  // 原子操作读取缓存并判断版本
  const cacheData = redis.get(cacheKey)
  if (!cacheData || event.clusterTime > cacheData.clusterTime) {
    const oldValue = cacheData?.value || null
    // 原子写入新值和新版本号
    redis.set(cacheKey, {value: event.fullDocument, clusterTime: event.clusterTime})
    // 后续业务逻辑使用oldValue
  }
  

路径B：基于副本集oplog自研监听（无缓存依赖） #

你提到的副本集方案可以解决上述问题，不需要额外缓存：

• 直接连接副本集的oplog集合（local.oplog.rs）监听所有写入操作，oplog里本身会记录更新前的操作日志，更新操作的o2字段存储了查询条件，o字段存储了更新操作的具体内容，删除操作的o字段存储了被删文档的_id；
• 当监听到更新/删除操作时，直接根据oplog里的文档id，从一个延迟节点（比如延迟10s的副本集节点）查询该文档的状态，拿到的就是变更前的旧值；
• 该方案的优势是完全不需要维护额外缓存，没有同步问题，缺陷是需要自行解析oplog格式，还要维护延迟节点，复杂度比缓存方案高。

内容的提问来源于stack exchange，提问作者Igor