Node.js - 使用回调插入 MongoDB

Question

我正在尝试使用async 的waterfall 方法将一些数据插入到我的MongoDB 集合中，以确保一切都以正确的顺序完成。最初我的数据以错误的顺序插入 MongoDB，过了一段时间我意识到这是因为我的 insert 是异步运行的，所以我无法保证数据会以正确的方式保存顺序。

这是我的async.waterfall：

async.waterfall([
    function(callback) {
      request.get(apiUrl, function(error, response, body) {
        if (!error && response.statusCode == 200) {
          data = JSON.parse(body);
          callback(error, data.items);
        }
      });
    },
    function(data, callback) {
      removeInvalidItems(data);
      var items = convertToObj(data);
      callback(null, items);
    },
    function(items, callback) {
      console.log(items);
      // Insert to collection here
    }
], function (err, result) {
    // do something here
});

因此，在瀑布的最后一个函数中，我已将items 记录到控制台，它以正确的顺序显示数据。所以现在，我需要做的就是将数据插入到我的集合中，并在回调中关闭连接并移出瀑布到最终的function。够容易吧？好吧，这就是我现在正在尝试的：

  MongoClient.connect(config.db, function(err, db) {
    if (err) {
      console.log(err);
    } else {
      console.log("Connected correctly to server");

      db.collection('items').insert(items, function(err, result) {
        if (err) {
          console.log(err);
        }
        db.close();
        callback(null, 'done');
      });
    }
  });

db.close() 和 callback(null, 'done') 在收到回调后应该关闭连接并移出瀑布。然而，我似乎得到的只是RangeError: Maximum call stack size exceeded。我已经尝试了所有我能想到的方法，但我无法让这个插入与回调一起工作并移出瀑布。

谁能帮忙？

Answer 1

我找到了我自己的问题here 的答案。问题是当我将数据批量插入 MongoDB 时，我收到了 Maximum call stack size exceeded 错误。我在运行db.collection('something').insert(data, function().... 时使用的数据数组是Users 的数组，根据称为User 的Mongoose 模式，因此data 数组中的每个项目都创建为var item = new User({});。显然，MongoDB 不喜欢这样。我认为这是在生产中插入数据的正确方法，这就是我使用数据的方式（使用User 对象）。

为了纠正这个问题，我只是用var item = {} 替换了var item = new User({})，这样我只是在创建一个新的JavaScript 对象。这工作正常，并以正确的顺序将数据插入数据库，没有错误。感谢写其他答案的人，我用这些来修复和改进我的回调！

Answer 2

应该为 async.waterfall() 的每个任务调用所有回调。它仅在您的第一个瀑布任务成功的情况下被调用......并且在第三个瀑布任务中根本不会被调用。

这两个调用没有回调：

removeInvalidItems(data);
items = convertToObj(data);

所以我假设它们不是异步的。而且由于它们不是异步的，因此您不必将它们作为 async.waterfall() 的单独任务。所以这里是修改后的代码（我假设你的数据转换是正确的）：

async.waterfall([
    function(callback) {
        request.get(apiUrl, function(error, response, body) {
            var data;
            if (!error && response.statusCode == 200) {
                data = JSON.parse(body);
                return callback(null, data.items);
            }
            callback(error || response); // error or statusCode !== 200
        });
    },
    function(data, callback) {
        var items;

        // There is no callback for these, so I'm assuming they are not asynchronous
        removeInvalidItems(data);
        items = convertToObj(data);

        MongoClient.connect(config.db, function(err, db) {
            if (err) {
                console.log(err);
                return callback(err);
            }
            console.log("Connected correctly to server");
            db.collection('items').insert(items, function(err, result) {
                if (err || !result) {
                    console.log(err);
                    db.close();
                    return callback(err || {message: 'There is a problem inserting.'});
                }
                db.close();
                callback(null);
            });
        });
    }
], function (err, result) {
    if (err) {
        console.log(err);
    } else {
        console.log('done');
    }
});

Answer 3

这个项目集合的大小是多少？它可以超过调用堆栈的大小吗？要获得您当前的大小运行：

node --v8-options | grep -B0 -A1 stack_size

默认大小为 492 kB（32 位）和 984 kB（64 位）。

要增加调用堆栈大小，请尝试使用参数启动节点：

node --max-stack-size={your-desired-size-in-bytes}

考虑到节点可能没有时间在每个瀑布回调之间清除调用堆栈。尝试在 nextTick 上运行回调：

async.waterfall([
    function(callback) {
      request.get(apiUrl, function(error, response, body) {
        if (!error && response.statusCode == 200) {
          data = JSON.parse(body);
          callback(error, data.items);
        }
      });
    },
    function(data, callback) {
      removeInvalidItems(data);
      var items = convertToObj(data);
      process.nextTick(function() {   // callback on next tick
          callback(null, items);
      });
    },
    function(items, callback) {
      console.log(items);
      // Insert to collection here
    }
], function (err, result) {
    // do something here
});