node.js - 使用 Mongoose 在 MongoDB 中更新许多记录的正确方法是什么

Question

我正在使用 Mongoose 从 MongoDB 中提取一些记录，将它们导入另一个系统，然后我想将所有这些文档的状态(文档属性)设置为 已处理。

我可以找到这个解决方案:Update multiple documents by id set. Mongoose

我想知道这是否是正确的方法，建立一个包含所有文档 ID 的标准，然后执行更新。还请考虑一个事实，即它将是许多文件。

(更新查询的限制是多少？到处都找不到。官方文档:http://mongoosejs.com/docs/2.7.x/docs/updating-documents.html)

Answer 1

建立一个包含所有文档 id 的标准然后执行更新的方法必然会导致潜在的问题。当您使用每个文档迭代发送更新操作的文档列表时，在 Mongoose 中，您冒着炸毁服务器的风险，尤其是在处理大型数据集时，因为您无需等待异步调用完成，然后再继续下一个迭代。您将基本上构建一个未解决操作的“堆栈”，直到这导致问题 - Stackoverflow。

例如，假设您有一个文档 ID 数组，您想要更新状态字段上的匹配文档:

const processedIds = [
  "57a0a96bd1c6ef24376477cd",
  "57a052242acf5a06d4996537",
  "57a052242acf5a06d4996538"
];

您可以在哪里使用 updateMany()方法

Model.updateMany(
  { _id: { $in: processedIds } }, 
  { $set: { status: "processed" } }, 
  callback
);

或者对于非常小的数据集，您可以使用 forEach() 方法对数组进行迭代并更新您的集合:

processedIds.forEach(function(id)){
  Model.update({ _id: id}, { $set: { status: "processed" } }, callback);
});

上述方法适用于小型数据集。但是，当您面对成千上万个要更新的文档时，这会成为一个问题，因为您将在循环中重复服务器调用异步代码。

要克服这个问题，请使用 async 的 eachLimit 并遍历数组，为每个项目执行 MongoDB 更新操作，同时永远不会同时执行超过 x 个并行更新。

---

最好的方法是为此使用批量 API，这在批量处理更新方面非常有效。性能与对众多文档中的每一个都调用更新操作的区别在于，批量 API 不是在每次迭代时将更新请求发送到服务器，而是在每 1000 个请求(批处理)中发送一次请求。

对于支持 MongoDB Server 3.2.x 的 Mongoose 版本 >=4.3.0，您可以使用 bulkWrite() 更新。下面的例子展示了如何去做:

const bulkUpdateCallback = function(err, r){
  console.log(r.matchedCount);
  console.log(r.modifiedCount);
}

// Initialize the bulk operations array
const bulkUpdateOps = [], counter = 0;

processedIds.forEach(function (id) {
  bulkUpdateOps.push({
    updateOne: {
      filter: { _id: id },
      update: { $set: { status: "processed" } }
    }
  });
  counter++;

  if (counter % 500 == 0) {
    // Get the underlying collection via the Node.js driver collection object
    Model.collection.bulkWrite(bulkUpdateOps, { ordered: true, w: 1 }, bulkUpdateCallback);
    bulkUpdateOps = []; // re-initialize
  }
})

// Flush any remaining bulk ops
if (counter % 500 != 0) {
  Model.collection.bulkWrite(bulkOps, { ordered: true, w: 1 }, bulkUpdateCallback);
}

---

对于支持 MongoDB Server >=2.6 的 Mongoose 版本 ~3.8.8、~3.8.22、4.x .x，你可以使用 Bulk API 如下

var bulk = Model.collection.initializeOrderedBulkOp(),
    counter = 0;

processedIds.forEach(function(id) {
    bulk.find({ "_id": id }).updateOne({ 
        "$set": { "status": "processed" }
    });

    counter++;
    if (counter % 500 == 0) {
        bulk.execute(function(err, r) {
           // do something with the result
           bulk = Model.collection.initializeOrderedBulkOp();
           counter = 0;
        });
    }
});

// Catch any docs in the queue under or over the 500's
if (counter > 0) {
    bulk.execute(function(err,result) {
       // do something with the result here
    });
}