插入许多文档后关闭node.js中的mongodb连接

Question

这个问题已经被问到,但OP接受的答案没有满足我的特殊需求.

在插入大量数据时关闭node.js中的mongodb连接

我有一个实用程序脚本,可以为多个集合添加大量记录.实际上它只是一个导入,使用byline读取非常大的文本文件,然后将数据插入到集合中:

var MongoClient = require("mongodb").MongoClient;
var fs = require("fs");
var byline = require("byline");

var inStream = fs.createReadStream("data.txt", { encoding: "utf8" });

var byLineStream = byline.createStream(inStream);

MongoClient.connect("mongodb://localhost:27017/test", { native_parser: true}, function(err, db) {  
    var collection = db.collection("Data");
    db.dropCollection("Data", function(err, result) {
        byLineStream.on("data", function(line) {
            var o = parseLineToObject(line);
            collection.insert(o);
        });
    });
});

建议的答案是将所有数据推送到一个数组中,然后在完成后使用单个write和一个回调来关闭数据库.这不是一个好的答案,因为我正在使用的文件非常大,因此消耗大量内存.

提出类似问题的另一个解决方案是使用async包创建一个函数数组,然后并行运行它们.另一个半身像,但至少它不会创造一个巨大的单一插入.

所以问题是:如果所有插入完成后如何关闭MongoDB连接,以便我的脚本退出并且不挂起？

我应该补充一点,我已经尝试了计数方法,我在insert回调中增加了一个计数器变量.它不起作用,因为在插入中的某个时刻,回调执行和完成的速度比插入完成的速度快,导致计数器在插入仍然继续时命中0,从而关闭数据库.

Answer 1

您应该在读取所有行时设置标志:

var readAllLines = false;

byLineStream.on("end", function() {
  readAllLines = true;
});

接下来,在插入每条记录后检查该标志.但是,您还需要跟踪已读取的行数以及插入的行数,因此如果已插入所有行(甚至不按顺序),则只关闭数据库.

把所有东西放在一起:

db.dropCollection("Data", function(err, result) {
  var lineCount     = 0;
  var readAllLines  = false;

  byLineStream.on("end", function() {
    readAllLines = true;
  });

  byLineStream.on("data", function(line) {
    lineCount++;
    var o = parseLineToObject(line);
    collection.insert(o, { w : 1 }, function() {
      if (--lineCount === 0 && readAllLines) {
        // we've read and inserted all lines
        db.close();
      }
    });
  });
});

但是,我确实认为将回调传递给insert('安全模式')比你当前的解决方案要慢,insert但是你不要等待它的结果.为了加快速度,您可以在将它们插入一个语句之前缓冲X行数量,而不是单独写入每一行.

与此类似的东西(没有行计数):

var buffer = [];
byLineStream.on("data", function(line) {
  buffer.push(parseLineToObject(line));
  if (buffer.length > 100 || readAllLines) {
    collection.insert(buffer, { w : 1 }, function() {
      if (readAllLines) {
        db.close();
      }
    });
    buffer = [];
  }
});