Mongo Change Stream如何获取事件发生时间戳？

别担心，Change Stream完全支持获取事件对应的时间戳，而且针对脚本中断后用resume token恢复的场景，也有对应的处理方式。我给你梳理几个实用的方案：

• 直接读取事件自带的clusterTime字段
  每个Change Stream返回的事件文档里都内置了clusterTime字段，它和oplog里的timestamp完全对应，代表事件发生时的MongoDB集群时间。这个字段是官方提供的标准字段，所有类型的Change Stream事件（插入、更新、删除等）都会包含它。

  举个Node.js驱动的使用例子：

  const { MongoClient } = require('mongodb');
  
  async function watchCollection() {
    const client = new MongoClient('mongodb://localhost:27017');
    await client.connect();
    const collection = client.db('test').collection('data');
    
    const changeStream = collection.watch();
    for await (const changeEvent of changeStream) {
      // 获取集群时间戳并转换为可读日期
      const clusterTs = changeEvent.clusterTime;
      const eventTime = new Date(clusterTs.getHighBits() * 1000 + clusterTs.getLowBits() / 1000);
      console.log(`事件发生时间: ${eventTime.toISOString()}`);
      
      // 这里处理你的业务逻辑
      console.log('事件详情:', changeEvent);
    }
  }
  
  watchCollection().catch(console.error);
  

• 从Resume Token中解析恢复点时间戳
  如果脚本中断后需要用resume token恢复监听，你可以从resume token里解析出对应的时间戳，明确知道恢复的起始时间点。Resume Token是一个BSON文档，其中的_data字段包含了时间戳等元数据（不同MongoDB版本格式略有差异，但核心逻辑一致）。

  比如用Python的pymongo驱动解析的示例：

  from bson import timestamp
  from pymongo import MongoClient
  
  client = MongoClient('mongodb://localhost:27017')
  db = client['test']
  collection = db['data']
  
  # 假设你之前保存了中断时的resume token
  saved_resume_token = # 你的resume token变量
  
  # 解析resume token中的时间戳
  if '_data' in saved_resume_token:
      # 提取前8字节作为时间戳的二进制数据
      ts_bytes = saved_resume_token['_data'][:8]
      event_ts = timestamp.Timestamp.from_bytes(ts_bytes)
      print(f"恢复点对应的时间戳: {event_ts}")
      print(f"恢复点对应的可读时间: {event_ts.as_datetime()}")
  
  # 用resume token恢复监听
  change_stream = collection.watch(resume_after=saved_resume_token)
  for change in change_stream:
      # 处理恢复后的事件
      pass
  

• 自定义时间戳字段（可选补充方案）
  如果你需要更灵活的时间控制（比如业务层的时间记录），可以在写入数据时手动添加时间戳字段，用MongoDB的$currentDate操作符自动设置当前时间。这样在Change Stream事件中，你可以直接读取这个自定义字段，它和事件发生时间几乎同步（极端延迟情况除外）。

  插入文档时的示例：

  db.data.insertOne({
    "content": "sample content",
    "createdAt": { "$currentDate": { "$type": "timestamp" } },
    "updatedAt": { "$currentDate": { "$type": "timestamp" } }
  })
  

  更新文档时可以同步更新updatedAt：

  db.data.updateOne(
    { "_id": ObjectId("...") },
    { 
      "$set": { "content": "updated content" },
      "$currentDate": { "updatedAt": true }
    }
  )
  

  之后在Change Stream事件中，就能通过changeEvent.fullDocument.createdAt或changeEvent.fullDocument.updatedAt获取时间戳。

以上几种方案里，clusterTime是最推荐的官方原生方案，完全不需要依赖直接监听oplog，同时也能完美适配resume token恢复的场景。

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