如何判断MongoDB中$merge聚合阶段已完成？

我有一个以$merge阶段结尾的聚合管道，代码如下：

db.collection("mycoll").aggregate([..., {$merge:{into:"myview"}}])

我需要在运行时生成物化视图，并随数据变更保持其更新。当前遇到的问题是：若$merge尚未完成就对目标集合执行更新，仅会更新已合并的文档，其余文档会以过时值插入。为解决此问题，我需要在$merge运行时延迟或阻止更新，但无法获知$merge何时完成。此外，使用$merge时聚合返回的游标为空，无法通过游标追踪进度；查看currentOp也仅显示$merge执行的单个文档更新操作，看不到带$merge的聚合任务。请问如何判断$merge已完成？

1. 同步等待聚合执行完成 #

大多数MongoDB驱动的聚合方法支持同步阻塞调用（或通过异步语法等待完成），即使$merge返回空游标，也会等待整个聚合流程（包括$merge阶段的所有写入操作）彻底结束后才继续后续代码。

以Node.js驱动为例，使用await等待聚合Promise完成：

await db.collection("mycoll").aggregate([..., {$merge:{into:"myview"}}]).toArray();
// 后续更新操作放在这里，确保在$merge完成后执行

这种方式无需额外状态追踪，直接利用语言层面的异步等待机制保证执行顺序。

2. 自定义状态标记实现并发控制 #

如果必须异步执行聚合，可以在目标集合中添加一个专用的状态文档，用来标记$merge的运行状态：

• 聚合启动前，原子性写入状态：

  db.myview.updateOne(
    {_id: "merge_status"},
    {$set: {running: true}},
    {upsert: true}
  )
  

• 聚合完成后（同步等待结束后），更新状态为完成：

  db.myview.updateOne({_id: "merge_status"}, {$set: {running: false}})
  

• 后续更新操作前，先原子性检查状态：

  const status = db.myview.findOne({_id: "merge_status"});
  if (!status || !status.running) {
    // 执行更新操作
  }
  

注意：高并发场景下建议使用findAndModify实现原子性检查+更新，避免竞态条件。

3. 精准过滤currentOp中的聚合任务 #

你之前看不到完整聚合任务，是因为currentOp默认展示的细节不足。使用带过滤条件的currentOp查询，可以定位到包含$merge阶段的聚合任务：

db.currentOp({
  "command.aggregate": "mycoll",
  "command.pipeline": {$elemMatch: {$merge: {$exists: true}}}
})

如果查询结果为空，说明$merge对应的聚合任务已经完成；如果返回结果，则表示聚合仍在运行。

4. 改用MongoDB原生物化视图 #

从MongoDB 4.2开始，官方支持自动刷新的物化视图，无需手动维护$merge流程，原生保证视图一致性：

• 创建自动刷新的物化视图：

  db.createMaterializedView(
    "myview",
    "mycoll",
    [/* 你的聚合管道（无需$merge阶段） */],
    {refreshInterval: {enabled: true, interval: 3600}} // 每小时自动刷新一次
  )
  

• 按需手动刷新时，调用refreshMaterializedView命令，该命令会阻塞直到刷新完成：

  db.runCommand({refreshMaterializedView: "myview"})
  // 后续更新操作放在命令返回后执行
  

内容的提问来源于stack exchange，提问作者Jean-Samuel Girard