node.js - 如何更新 Mongoose 中的记录

Question

我试图在循环中更新记录的一个字段，注意该字段是一个数组，我的代码如下，

  Employeehierarchy = mongoose.model('Employeehierarchy'),
  function (done) {
      var ObjectId = require('mongoose').Types.ObjectId; 
      var item = {'childrens':{$in:[ ObjectId(employee.manager)]}};
      Employeehierarchy.find(item).exec(function (err, employeehierarchy) {
        if (err) {
          return res.status(400).send({ message: errorHandler.getErrorMessage(err) });
        } else {
          if (employeehierarchy && employeehierarchy.length > 0) {
              employeehierarchy.forEach(function (v1, i1) {
              v1.parents = employee._id;
              employeehierarchy = _.extend(employeehierarchy, v1);
              employeehierarchy.save(function (err) {
              });
            }); done();
          } else {
            done();
          }
        }
      });
    },

我的模式，

var EmployeehierarchySchema = new Schema({
  name: {
    type: String,
    default: ''
  },
  parents: {
    type: Array,
    default: ''
  },
  childrens: {
    type: Array,
    default: ''
  },
});

我不知道哪里出错了，谁能帮帮我？

Answer 1

您可以使用 Bulk Write Operations用于更新模型的 API。但是为了使用底层批量操作 API，您应该通过 mongoose 模型的 .collection 属性访问它，并且在使用 API 之前，等待 mongoose 成功连接到数据库，因为 Mongoose 没有尚不真正支持“initializeOrderedBulkOp()”函数，因为它不适用于 mongoose 的内部缓冲系统。

您可以实现类似下面的内容，它使用 Promises 来处理 node.js 中批量 API 的异步特性。

模型声明

var mongoose = require('mongoose'),
    express = require('express'),
    Promise = require('bluebird'),
    Schema = mongoose.Schema;

var employeeHierarchySchema = new Schema({
    name: {
        type: String,
        default: ''
    },
    parents: {
        type: Array,
        default: ''
    },
    childrens: {
        type: Array,
        default: ''
    }
});

var Employeehierarchy = mongoose.model('Employeehierarchy', employeeHierarchySchema);    

使用 Promises 执行批量更新的函数:

function bulkUpdate(Model, query){    
    return new Promise(function(resolve, reject){
        var ops = [],
            collection = Model.collection;

        Model.find(query).lean().exec(function (err, docs) {
            if (err) return reject(err);

            docs.forEach(function (doc){
                ops.push({
                    "updateOne": {
                        "filter": { "_id": doc._id },
                        "update": {
                            "$push": { "parents": doc._id }
                        }
                    }
                });

                if (ops.length === 500) {
                    collection.bulkWrite(ops, function(err, result) {
                        if (err) return reject(err);                        
                        ops = [];
                        resolve(result);
                    });
                }                       
            });     

            if (ops.length > 0) {            
                collection.bulkWrite(ops, function(err, result) {
                    if (err) return reject(err);
                    resolve(result);
                });         
            }           
        });     
    });
}

旧 MongoDB 版本的替代函数，使用 initializeUnorderedBulkOp()

function bulkUpdate(Model, query){    
    return new Promise(function(resolve, reject){
        var bulk = Model.collection.initializeUnorderedBulkOp(),
            counter = 0;

        Model.find(query).lean().exec(function (err, docs) {
            if (err) return reject(err);

            docs.forEach(function (doc){
                counter++;

                bulk.find({ "_id": doc._id }).updateOne({
                    "$push": { "parents": doc._id }
                });

                if (counter % 500 == 0 ) {
                    bulk.execute(function(err, result) {
                        if (err) return reject(err);                        
                        bulk = Model.collection.initializeUnorderedBulkOp();                        
                        resolve(result);
                    });
                }                       
            });     

            if (counter % 500 != 0 ) {            
                bulkUpdateOps.execute(function(err, result) {
                    if (err) return reject(err);
                    resolve(result);
                });         
            }           
        });     
    });
}

连接MongoDB的函数

function connect(uri, options){
    return new Promise(function(resolve, reject){
        mongoose.connect(uri, options, function(err){
            if (err) return reject(err);
            resolve(mongoose.connection);
        });
    });
}   

在连接上运行批量更新

connect('mongodb://localhost/yourdb', {}).then(function(db){
    var query = { "childrens": employee.manager };
    bulkUpdate(Employeehierarchy, query).then(function(res){
        console.log('Bulk update complete.', res);      
    }, function(err){
        res.status(400).send({ message: errorHandler.getErrorMessage(err) });
        db.close();
    });
}, function(err){
    res.status(400).send({ message: errorHandler.getErrorMessage(err) });
});