4. 变更事件文档结构详解

Change Streams 返回的每一个变更事件，本质上都是一份 BSON 文档。它不像普通查询结果那样只关心业务字段，而是把"这次变更发生在哪个集合、对哪条文档、做了什么操作、操作前后的样子、发生在什么时间点"这些元信息全部封装在一个 JSON 结构中。理解这份文档的字段语义，是后续做事件路由、幂等去重、增量同步和审计回放的基础。本节以 MongoDB 6.0 的默认输出为例，逐字段拆解一份典型的 insert 与 update 事件。

一份完整的变更事件长什么样

先看一条插入事件的实际结构（示例数据，非实测结果）：

{
  "_id": { "_data": "8264F2A1B2000000012B022C0100296E5A1004..." },
  "operationType": "insert",
  "clusterTime": Timestamp(1710392000, 1),
  "wallTime": ISODate("2024-03-14T08:53:20.000Z"),
  "fullDocument": {
    "_id": ObjectId("65f2a1b2c3d4e5f6a7b8c9d0"),
    "orderNo": "SO20240314001",
    "amount": 1280.00,
    "status": "created"
  },
  "ns": { "db": "shop", "coll": "orders" },
  "documentKey": { "_id": ObjectId("65f2a1b2c3d4e5f6a7b8c9d0") }
}

可以看到，事件文档由若干固定字段组成，其中 _id、operationType、clusterTime、ns、documentKey 几乎在所有事件类型中都会出现，而 fullDocument、updateDescription、to 等字段则随操作类型变化。下面逐个说明。

_id：事件的恢复令牌，不是业务主键

事件顶层的 _id 是一个形如 { "_data": "..." } 的对象，它被称为 resume token（恢复令牌）。它的作用不是标识业务文档，而是标识"这条变更在 oplog 中的位置"。当消费者断线重连时，把上一次成功处理事件的 _id 传给 resumeAfter 或 startAfter，MongoDB 就能从该位置之后继续投递事件，实现"至少一次"的续传语义。

operationType：决定后续字段如何解析

operationType 是一个字符串枚举，常见取值包括：

• insert：插入文档，携带 fullDocument。
• update：更新文档，携带 updateDescription，是否带 fullDocument 取决于配置。
• replace：整文档替换，携带 fullDocument。
• delete：删除文档，只有 documentKey，没有 fullDocument。
• drop、rename、dropDatabase、invalidate：集合或库级事件，通常没有 documentKey。

消费端的第一层分支判断几乎总是基于 operationType：如果是 delete，就只能靠 documentKey 定位被删文档；如果是 update，则要读 updateDescription 才能知道改了哪些字段。

ns 与 documentKey：定位"改的是谁"

ns 是 namespace 的缩写，包含 db 和 coll 两个子字段，指明变更发生在哪个数据库的哪个集合。对于跨集合监听（如监听整个库），消费端必须用 ns.coll 做路由分发。

documentKey 则给出被修改文档的主键，默认就是 { "_id": ... }。它是幂等处理的关键：当同一条事件因重试被重复消费时，用 documentKey._id 作为去重键即可。注意，如果集合使用了分片键且分片键不是 _id，documentKey 中可能同时包含分片键字段。

fullDocument：操作后的完整文档快照

对于 insert 和 replace，fullDocument 默认就是操作后的完整文档。对于 update，默认情况下 fullDocument 并不存在，只有当你显式配置了 fullDocument: "updateLookup" 时，MongoDB 才会额外去集合中查询一次，把更新后的文档补进事件里。

db.orders.watch([], { fullDocument: "updateLookup" })

updateDescription：字段级变更明细

当 operationType 为 update 时，事件会携带 updateDescription，其结构如下：

"updateDescription": {
  "updatedFields": { "status": "paid", "amount": 1380.00 },
  "removedFields": ["couponCode"],
  "truncatedArrays": [
    { "field": "tags", "newSize": 5 }
  ]
}

• updatedFields：被设置或修改的字段及其新值，支持点号路径（如 "address.city"）。
• removedFields：被 $unset 删除的字段名数组。
• truncatedArrays：当使用 $push、$pop 等数组操作且数组被截断时，记录字段名和截断后的新长度。

有了这三个子字段，消费端就可以做"增量合并"：把 updatedFields 覆盖到本地副本，把 removedFields 从本地副本删除，而不必每次都拉取完整文档。

clusterTime 与 wallTime：时间语义要分清

clusterTime 是一个 BSON Timestamp，由"秒级时间戳 + 递增序号"组成，代表该操作在集群逻辑时钟中的位置。它单调递增，适合用来做事件排序和"处理到哪个时间点"的进度标记。而 wallTime 是操作实际发生的墙上时间（ISODate），更接近人类理解的时间，但可能因时钟漂移而不完全可靠。

实践中的常见做法是：用 clusterTime 做严格排序与断点续传，用 wallTime 做业务侧的时间展示与审计记录。两者不要混用。

字段组合与消费策略对照

把上述字段按操作类型组合起来，可以整理出一张消费端最常用的判断表：

一个可落地的解析示例

下面这段 Node.js 片段展示了如何根据 operationType 分派处理，并把恢复令牌写入检查点（示例代码，参数为教学假设值）：

const changeStream = db.collection('orders').watch([], {
  fullDocument: 'updateLookup'
});

for await (const event of changeStream) {
  const { operationType, documentKey, fullDocument, updateDescription, ns, clusterTime } = event;

  switch (operationType) {
    case 'insert':
    case 'replace':
      await upsertToWarehouse(ns.coll, documentKey._id, fullDocument);
      break;
    case 'update':
      await applyPatch(ns.coll, documentKey._id, updateDescription);
      break;
    case 'delete':
      await removeFromWarehouse(ns.coll, documentKey._id);
      break;
    default:
      console.warn('未处理的集合级事件:', operationType, ns);
  }

  // 处理成功后持久化恢复令牌
  await saveCheckpoint(event._id, clusterTime);
}

这段代码体现了三个要点：第一，用 operationType 做分派；第二，用 documentKey._id 做幂等键；第三，用事件顶层的 _id 做检查点。只要这三件事做对，即使消费者重启或网络抖动，也能保证事件不丢、不重复生效。