【问题标题】:async.waterfall is not working with mongoose asynchronous functionsasync.waterfall 不适用于猫鼬异步函数
【发布时间】:2017-12-19 00:19:53
【问题描述】:
import Student from './api/user/student.model';
import Class from './api/user/class.model';
import User from './api/user/user.model';
import request from 'request';
import async from 'async';

const EventEmitter = require('events');

class MyEmitter extends EventEmitter {}

const myEmitter = new MyEmitter;


function seedDb() {
    Student.remove({}, (err, data) => {
        Class.remove({}, (err, data1) => {
            User.remove({}, (err, data2) => {
                User.create({
                    email: 'admin@example.com',
                    password: 'admin'
                }, (err, data3) => {
                    User.findOne({
                        email: 'admin@example.com'
                    }, (err, foundUser) => {
                        foundUser.save((err, done) => {
                            let url = 'https://gist.githubusercontent.com/relentless-coder/b7d74a9726bff8b281ace936757953e5/raw/6af59b527e07ad3a589625143fb314bad21d4f8e/dummydata.json';
                            let options = {
                                url: url,
                                json: true
                            }
                            request(options, (err, res, body) => {
                                if (!err && body) {
                                    let classId;
                                    async.forEachOf(body, (el, i) => {
                                        console.log(i);
                                        if (i % 4 === 0) {
                                            async.waterfall([(fn) => {
                                                Student.create(el, (err, success) => {
                                                    fn(null, success._id);
                                                })
                                            }, (id, fn) => {
                                                console.log('The class creation function is called');
                                                Class.create({name: `Class ${i}`}, (err, newClass)=>{
                                                    classId = newClass._id;
                                                    newClass.students.push(id);
                                                    newClass.save()
                                                    fn(null, 'done');
                                                })
                                            }])
                                        } else {
                                            async.waterfall([(fn) => {
                                                Student.create(el, (err, success) => {
                                                    fn(null, success._id)
                                                })
                                            }, (id, fn) => {
                                                console.log('Class find function is called and classId', id, classId)
                                                Class.findById(classId, (err, foundClass) => {
                                                    console.log(foundClass)
                                                    foundClass.students.push(id);
                                                    foundClass.save();
                                                    fn(null, 'done');
                                                })

                                            }])
                                        }
                                    })

                                }
                            });
                        })

                    });
                })
            })

        })


    })

}

我想要做的是,我有 20 个学生的样本数据。我必须将分布在 5 个班级的 20 名学生作为我的数据库的种子,每个班级有 4 名学生。

阻止我实现这一目标的是某些迭代,classId 的值是undefined

谁能帮我解决这个问题?

【问题讨论】:

  • 你为什么要这么做?相反,您应该简单地将 fn 作为“回调”参数提供给 mongoose 方法,就像在 (fn) => Student.create(el,fn), (success,fn) => // address as success._id here 中一样。您正在包装不需要包装的东西。坦率地说,瀑布“之前”的代码在这里更令人担忧。您的回调用法可以被清除,甚至这里使用 asyncjs 有点“老帽子”,可以很容易地用 Promise 来处理。但这里的查询数量也是惊人的,你的建模确实可以做一些工作。
  • 您能解释一下方法吗?您的评论对我应该使用什么其他方法并没有真正的帮助。

标签: node.js mongodb mongoose async.js


【解决方案1】:

现代异步/等待版本

在没有看到实际架构和预期结果的情况下,你并不完全清楚你在追求什么,但我可以提供一个近乎复制的内容,希望你可以遵循并从中学习。

正如评论中所述,您目前正在做的事情有几处错误以及过时的概念。但作为解析输入 url 内容并将其播种到数据库的一般目标,我建议采用这样的方法:

const request = require('request-promise-native'),
      mongoose = require('mongoose'),
      Schema = mongoose.Schema;

mongoose.Promise = global.Promise;
mongoose.set('debug',true);

const input = 'https://gist.githubusercontent.com/relentless-coder/b7d74a9726bff8b281ace936757953e5/raw/6af59b527e07ad3a589625143fb314bad21d4f8e/dummydata.json';

const uri = 'mongodb://localhost/school',
      options = { useMongoClient: true };

const studentSchema = new Schema({
  "student_id": { type: Number, unique: true },
  "student_name": String,
  "student_email": String,
  "neuroticism": Number,
  "extraversion": Number,
  "openness_to_experience": Number
});

const classSchema = new Schema({
  "course_number": { type: Number, unique: true },
  "course_name": String,
  "teacher_name": String,
  "teacher_number": Number,
  "students": [{ type: Schema.Types.ObjectId, ref: 'Student' }]
});


const Student = mongoose.model('Student', studentSchema);
const Class = mongoose.model('Class', classSchema);

function log(data) {
  console.log(JSON.stringify(data,undefined,2))
}

function extractPaths(model) {
  return Object.keys(model.schema.paths).filter( p =>
    ['_id','__v'].indexOf(p) === -1
  );
}

(async function() {

  try {
    console.log('starting');
    const conn = await mongoose.connect(uri,options);
    console.log('connected');

    // Clean models
    await Promise.all(
      Object.keys(conn.models) // <-- is the same thing
      //['Student','Class','User']
        .map( m => conn.models[m].remove({}) )
    );
    console.log('cleaned');

    let response = await request({ uri: input, json: true });

    for (let res of response) {
      let student = extractPaths(Student)
        .reduce((acc,curr) => Object.assign(acc,{ [curr]: res[curr] }),{});
      log(student);

      let sclass = extractPaths(Class).filter(p => p !== 'students')
        .reduce((acc,curr) => Object.assign(acc,{ [curr]: res[curr] }),{});
      log(sclass);

      let fstudent = await Student.findOneAndUpdate(
        { student_id: student.student_id },
        student,
        { new: true, upsert: true }
      );

      let fclass = await Class.findOneAndUpdate(
        { course_number: sclass.course_number },
        {
          $setOnInsert: sclass,
          $addToSet: { 'students': fstudent._id }
        },
        { new: true, upsert: true }
      );

    }

  } catch(e) {
    console.error(e)
  } finally {
    mongoose.disconnect()
  }

})();

这里有几个非常不同的概念需要注意,主要是async/await 语法的使用,因为我们可以在现代 nodejs 版本中使用它,它使代码更加简洁。

与您当前代码的第一个重大不同之处在于处理清理数据的.remove() 语句。因此,现在我们不再使用链接回调,而是让Promise.all 告诉我们所有操作何时完成:

    // Clean models
    await Promise.all(
      Object.keys(conn.models) // <-- is the same thing
      //['Student','Class','User']
        .map( m => conn.models[m].remove({}) )
    );

下一个重大变化是使用常规的for of 循环,因为我们将再次将来自其中包含的任何异步操作的响应转到await。我们可能会更花哨,并执行类似的Promise.all 来并行运行多个迭代,但这将作为示例。

由于我们await,然后我们会分别进行实际写入数据库的每个调用,在那里我们可以使用响应数据提供给下一个调用。因此,创建Student 的响应可以稍后反馈到Class

我们真正改变的另一件事是我们如何从提要中提取数据以创建每个对象,以及我们如何实际进行更新。

这改为使用.findOneAndUpdate() 并发出"upserts",我们基本上通过主键“查找”当前文档,在它不存在的地方我们“创建”一个新文档,或者以其他方式找到它我们只是用新信息“更新”。

这主要通过Class 模型进行演示,其中我们在"students" 数组上$addToSet 将提供的学生_id 值作为“更新”,并且它实际上不会两次创建同一个学生。两者都是因为Student 在基于它们自己的student_id 值进行处理时不会被复制,$addToSet 也不会允许对Student 的引用多次插入到Class 对象中,@987654347 @数组。

courses 集合的最终输出以及每个引用的学生是:

{
        "_id" : ObjectId("5968597490aa0ed4e5db1c92"),
        "course_number" : 101,
        "teacher_number" : 539224,
        "teacher_name" : "Merideth Merrill",
        "course_name" : "Physics 1",
        "__v" : 0,
        "students" : [
                ObjectId("5968597490aa0ed4e5db1c90"),
                ObjectId("5968597490aa0ed4e5db1c94"),
                ObjectId("5968597490aa0ed4e5db1c97"),
                ObjectId("5968597490aa0ed4e5db1c9a"),
                ObjectId("5968597490aa0ed4e5db1c9d"),
                ObjectId("5968597490aa0ed4e5db1ca0"),
                ObjectId("5968597490aa0ed4e5db1ca3"),
                ObjectId("5968597490aa0ed4e5db1ca6"),
                ObjectId("5968597490aa0ed4e5db1ca9"),
                ObjectId("5968597490aa0ed4e5db1cac")
        ]
}
{
        "_id" : ObjectId("5968597490aa0ed4e5db1cb1"),
        "course_number" : 102,
        "teacher_number" : 539224,
        "teacher_name" : "Merideth Merrill",
        "course_name" : "AP Physics C",
        "__v" : 0,
        "students" : [
                ObjectId("5968597490aa0ed4e5db1caf"),
                ObjectId("5968597490aa0ed4e5db1cb3"),
                ObjectId("5968597490aa0ed4e5db1cb6"),
                ObjectId("5968597490aa0ed4e5db1cb9"),
                ObjectId("5968597490aa0ed4e5db1cbc")
        ]
}
{
        "_id" : ObjectId("5968597590aa0ed4e5db1cc1"),
        "course_number" : 103,
        "teacher_number" : 731037,
        "teacher_name" : "Kelly Boyd",
        "course_name" : "English 11",
        "__v" : 0,
        "students" : [
                ObjectId("5968597590aa0ed4e5db1cbf"),
                ObjectId("5968597590aa0ed4e5db1cc3"),
                ObjectId("5968597590aa0ed4e5db1cc6"),
                ObjectId("5968597590aa0ed4e5db1cc9"),
                ObjectId("5968597590aa0ed4e5db1ccc")
        ]
}

当然,所有来自源的 Student 条目也包含在内并填充到它们自己的集合中。


所以我们通常在这里对一些技术进行现代化改造,结果是代码更加简洁,易于遵循逻辑,而且我们还通过简单地使用最有效的方式减少了很多与数据库的“来回”通信实际写入和读取数据的方法。处理每个项目时是.findOneAndUpdate()

当然,这不是您尝试实现的 100%,但它至少应该展示如何以您可以遵循和学习的方式更有效地实现它。


实际的 async.js 版本

如果在经历了所有这些之后,你仍然坚持使用 async.js,那么这个清单更正了用法:

const async = require('async'),
      request = require('request'),
      mongoose = require('mongoose'),
      Schema = mongoose.Schema;

mongoose.Promise = global.Promise;
mongoose.set('debug',true);

const input = 'https://gist.githubusercontent.com/relentless-coder/b7d74a9726bff8b281ace936757953e5/raw/6af59b527e07ad3a589625143fb314bad21d4f8e/dummydata.json';

const uri = 'mongodb://localhost/school',
      options = { useMongoClient: true };

const studentSchema = new Schema({
  "student_id": { type: Number, unique: true },
  "student_name": String,
  "student_email": String,
  "neuroticism": Number,
  "extraversion": Number,
  "openness_to_experience": Number
});

const classSchema = new Schema({
  "course_number": { type: Number, unique: true },
  "course_name": String,
  "teacher_name": String,
  "teacher_number": Number,
  "students": [{ type: Schema.Types.ObjectId, ref: 'Student' }]
});


const Student = mongoose.model('Student', studentSchema);
const Class = mongoose.model('Class', classSchema);

function log(data) {
  console.log(JSON.stringify(data,undefined,2))
}

function extractPaths(model) {
  return Object.keys(model.schema.paths).filter( p =>
    ['_id','__v'].indexOf(p) === -1
  );
}

async.series(
  [
    (callback) => mongoose.connect(uri,options,callback),

    // Clean data
    (callback) =>
      async.each(mongoose.models,(model,callback) =>
        model.remove({},callback),callback),


    (callback) =>
      async.waterfall(
        [
          (callback) => request({ uri: input, json: true },
            (err,res) => callback(err,res)),

          (response,callback) =>
            async.eachSeries(response.body,(res,callback) =>
              async.waterfall(
                [
                  (callback) => {
                    let student = extractPaths(Student)
                      .reduce((acc,curr) =>
                        Object.assign(acc,{ [curr]: res[curr] }),
                        {}
                      );
                    log(student);
                    Student.findOneAndUpdate(
                      { student_id: student.student_id },
                      student,
                      { new: true, upsert: true },
                      callback
                    );
                  },

                  (student,callback) => {
                    console.log(student);
                    let sclass = extractPaths(Class)
                      .filter(p => p !== 'students')
                      .reduce((acc,curr) =>
                        Object.assign(acc,{ [curr]: res[curr] }),
                        {}
                      );
                    log(sclass);
                    Class.findOneAndUpdate(
                      { course_number: sclass.course_number },
                      {
                        $setOnInsert: sclass,
                        $addToSet: { 'students': student._id }
                      },
                      { new: true, upsert: true },
                      callback
                    );
                  }
                ],
                callback
              ),
              callback
            )
        ],
        callback
      )
  ],
  (err) => {
    if (err) throw err;
    mongoose.disconnect();
  }
)

【讨论】:

  • 嗨,哇。您的回答教会了我很多东西,但我最终没有使用您的解决方案,因为它不断为虚拟数据中不存在的学生路径崩溃。很抱歉,非常感谢您抽出时间写下答案。我会继续研究它,因为我发现你的回答非常优雅。
  • @AyushBahuguna 我从来没有试图“为你编写代码”,实际上我特意这么说。这只是你应该如何做的一个“自包含的例子”。至于“崩溃”,这段代码工作得很好,并且产生的结果与包含的完全一样。所以不一定是你可以“拿起和使用”而不是打算成为的东西。但是你遇到的任何问题实际上都是你自己的。这里的代码完美运行。而是遵循示例并纠正您的错误。这就是目的。
  • @AyushBahuguna 抱歉到底是什么?没有人会为您编写代码。我们也不能,因为您没有提供完整的流程。你在这里有足够的信息来纠正你的错误并做一些真正适合你的事情。因此,您应该accept the answer 并根据给出的示例忙于纠正您自己的流程。您错误地执行了所有异步调用。现在您知道如何正确执行它们了。似乎这回答了我的问题。
【解决方案2】:

我想做的是,我有 20 个学生的样本数据。我必须将分布在 5 个班级的 20 名学生作为我的数据库的种子,每个班级有 4 名学生。

正如 Neil 所建议的,我们可以消除一些回调和循环。这个想法是根据功能分离代码。以下是我利用现有信息解决问题的方法。

'use strict';
let _ = require('lodash');
const BATCH_COUNT = 4;

function seedDb() {
  clearAll().then(() => {
    return Promisea.all([
      createSuperAdmin(),
      fetchStudentDetails()
    ]);
  }).then((result) => {
    let user = result[0];
    let body = result[1];
    return createStudents(body);
  }).then((users) => {
    let studentsBatch = groupByIds(_.map(users, '_id'), BATCH_COUNT);
    return addStudentsToClass(studentsBatch);
  }).then(() => {
    console.log('Success');
  }).catch((err) => {
    console.log('err', err.stack);
  });
}

function addStudentsToClass(batches) {
  let bulk = Class.collection.initializeOrderedBulkOp();
  for (let i = 0; i < _.size(batches); i++) {
    bulk.insert({
      name: `Class ${i}`,
      students: batches[i]
    });
  }
  return bulk.execute();
}

function createStudents(users) {
  let bulk = Student.collection.initializeOrderedBulkOp();
  _.each(users, (user) => {
    bulk.insert(user);
  });
  return bulk.execute();
}

function createSuperAdmin() {
  return User.findOneAndUpdate({
    email: 'admin@example.com',
    password: 'admin'
  }, {}, {
    new: true,
    upsert: true
  });
}

function groupByIds(ids, count) {
  let batch = [];
  let index = 0;
  while (index < ids.length) {
    let endIndex = index + count;
    batch.push(ids.slice(index, endIndex));
    index = endIndex;
  }
  return batch;
}

function fetchStudentDetails() {
  let options = {
    url: 'https://data.json', // Your URL
    json: true
  };
  return new Promise((resolve, reject) => {
    request(options, (err, res, body) => {
      if (err) {
        return reject(err);
      }
      return resolve(body);
    });
  });
}

function clearAll() {
  return Promise.all([
    Student.remove({}).exec(),
    Class.remove({}).exec(),
    User.remove({}).exec()
  ]);
}

【讨论】:

  • 嘿,谢谢。我让你的代码工作了,bulk.execute() 返回一个包含插入文档状态的对象,我在下面添加了一个新行,它将获取所有学生,然后将其传递给groupByIds 函数
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2018-08-30
  • 2015-10-23
  • 2021-11-08
  • 2021-06-18
  • 2013-06-24
  • 2020-11-16
相关资源
最近更新 更多