如何在 node.js 断开连接期间缓冲 MongoDB 插入？

Question

我们确实读取了一个包含大约 500k 元素的 XML 文件（使用 xml-stream），并将它们插入到 MongoDB 中，如下所示：

xml.on(`endElement: product`, writeDataToDb.bind(this, "product"));

插入writeDataToDb(type, obj) 看起来像这样：

collection.insertOne(obj, {w: 1, wtimeout: 15000}).catch((e) => { });

现在，当 Mongo 连接断开时，xml 流仍会读取，并且控制台会充斥着错误消息（无法插入、断开连接、EPIPE 损坏……）。

在docs 中写道：

当您关闭 mongod 进程时，驱动程序会停止处理操作并继续缓冲它们，因为 bufferMaxEntries 默认为 -1，这意味着缓冲所有操作。

这个缓冲区的实际作用是什么？

我们注意到，当我们插入数据并关闭 mongo 服务器时，这些东西被缓冲了，然后我们恢复了 mongo 服务器，本机驱动程序成功重新连接并且节点继续插入数据，但是缓冲的文档（在 mongo beeing 离线期间）做不再插入。

所以我质疑这个缓冲区及其用途。

目标：

我们正在寻找将插入保持在缓冲区中直到 mongo 返回（根据wtimeout 在 15000 毫秒内）的最佳方法，然后让插入缓冲的文档或使用我们尝试过的 xml.pause(); 和 xml.resume()成功。

基本上，我们需要一些帮助来处理断开连接而不会丢失数据或中断。

Answer 1

使用 insertOne() 插入 500K 元素是一个非常糟糕的主意。您应该改用bulk operations，它允许您在单个请求中插入许多文档。 （这里以 10000 为例，因此可以在 50 个单个请求中完成） 为避免缓冲问题，您可以手动处理：

1. 使用bufferMaxEntries: 0禁用缓冲
2. 设置重新连接属性：reconnectTries: 30, reconnectInterval: 1000
3. 创建一个 bulkOperation 并为其提供 10000 个项目
4. 暂停 xml 阅读器。尝试插入 10000 个项目。如果失败，则每 3000ms 重试一次，直到成功
5. 如果批量操作在执行过程中中断，您可能会遇到一些重复 ID 问题，请忽略它们（错误代码：11000）

这是一个示例脚本：

var fs = require('fs')
var Xml = require('xml-stream')

var MongoClient = require('mongodb').MongoClient
var url = 'mongodb://localhost:27017/test'

MongoClient.connect(url, {
  reconnectTries: 30,
  reconnectInterval: 1000,
  bufferMaxEntries: 0
}, function (err, db) {
  if (err != null) {
    console.log('connect error: ' + err)
  } else {
    var collection = db.collection('product')
    var bulk = collection.initializeUnorderedBulkOp()
    var totalSize = 500001
    var size = 0

    var fileStream = fs.createReadStream('data.xml')
    var xml = new Xml(fileStream)
    xml.on('endElement: product', function (product) {
      bulk.insert(product)
      size++
      // if we have enough product, save them using bulk insert
      if (size % 10000 == 0) {
        xml.pause()
        bulk.execute(function (err, result) {
          if (err == null) {
            bulk = collection.initializeUnorderedBulkOp()
            console.log('doc ' + (size - 10000) + ' : ' + size + ' saved on first try')
            xml.resume()
          } else {
            console.log('bulk insert failed: ' + err)
            counter = 0
            var retryInsert = setInterval(function () {
              counter++
              bulk.execute(function (err, result) {
                if (err == null) {
                  clearInterval(retryInsert)
                  bulk = collection.initializeUnorderedBulkOp()
                  console.log('doc ' + (size - 10000) + ' : ' + size + ' saved after ' + counter + ' tries')
                  xml.resume()
                } else if (err.code === 11000) { // ignore duplicate ID error
                  clearInterval(retryInsert)
                  bulk = collection.initializeUnorderedBulkOp()
                  console.log('doc ' + (size - 10000) + ' : ' + size + ' saved after ' + counter + ' tries')
                  xml.resume()
                } else {
                  console.log('failed after first try: ' + counter, 'error: ' + err)
                }
              })
            }, 3000) // retry every 3000ms until success
          }
        })
      } else if (size === totalSize) {
        bulk.execute(function (err, result) {
          if (err == null) {
            db.close()
          } else {
            console.log('bulk insert failed: ' + err)
          }
        })
      }
    })
  }
})

示例日志输出：

doc 0 : 10000 saved on first try
doc 10000 : 20000 saved on first try
doc 20000 : 30000 saved on first try
[...]
bulk insert failed: MongoError: interrupted at shutdown // mongodb server shutdown
failed after first try: 1 error: MongoError: no connection available for operation and number of stored operation > 0
failed after first try: 2 error: MongoError: no connection available for operation and number of stored operation > 0
failed after first try: 3 error: MongoError: no connection available for operation and number of stored operation > 0
doc 130000 : 140000 saved after 4 tries
doc 140000 : 150000 saved on first try
[...]

Answer 2

我不具体了解 Mongodb 驱动程序和这个条目缓冲区。也许它只保留特定场景的数据。

所以我将用一种更通用的方法来回答这个问题，该方法可以与任何数据库一起使用。

总而言之，你有两个问题：

1. 您没有从失败的尝试中恢复
2. XML 流发送数据太快

要处理第一个问题，您需要实现一个重试算法，以确保在放弃之前进行多次尝试。

要处理第二个问题，您需要对 xml 流实施背压。您可以使用pause 方法、resume 方法和输入缓冲区来实现。

var Promise = require('bluebird');
var fs = require('fs');
var Xml = require('xml-stream');

var fileStream = fs.createReadStream('myFile.xml');
var xml = new Xml(fileStream);

// simple exponential retry algorithm based on promises
function exponentialRetry(task, initialDelay, maxDelay, maxRetry) {
    var delay = initialDelay;
    var retry = 0;
    var closure = function() {
        return task().catch(function(error) {
            retry++;
            if (retry > maxRetry) {
                throw error
            }
            var promise = Promise.delay(delay).then(closure);
            delay = Math.min(delay * 2, maxDelay);
            return promise;
        })
    };
    return closure();
}

var maxPressure = 100;
var currentPressure = 0;
var suspended = false;
var stopped = false;
var buffer = [];

// handle back pressure by storing incoming tasks in the buffer
// pause the xml stream as soon as we have enough tasks to work on
// resume it when the buffer is empty
function writeXmlDataWithBackPressure(product) {
    // closure used to try to start a task
    var tryStartTask = function() {
        // if we have enough tasks running, pause the xml stream
        if (!stopped && !suspended && currentPressure >= maxPressure) {
            xml.pause();
            suspended = true;
            console.log("stream paused");
        }
        // if we have room to run tasks
        if (currentPressure < maxPressure) {
            // if we have a buffered task, start it
            // if not, resume the xml stream
            if (buffer.length > 0) {
                buffer.shift()();
            } else if (!stopped) {
                try {
                    xml.resume();
                    suspended = false;
                    console.log("stream resumed");
                } catch (e) {
                    // the only way to know if you've reached the end of the stream
                    // xml.on('end') can be triggered BEFORE all handlers are called
                    // probably a bug of xml-stream
                    stopped = true;
                    console.log("stream end");
                }
            }
        }
    };

    // push the task to the buffer
    buffer.push(function() {
        currentPressure++;
        // use exponential retry to ensure we will try this operation 100 times before giving up
        exponentialRetry(function() {
            return writeDataToDb(product)
        }, 100, 2000, 100).finally(function() {
            currentPressure--;
            // a task has just finished, let's try to run a new one
            tryStartTask();
        });
    });

    // we've just buffered a task, let's try to run it
    tryStartTask();
}

// write the product to database here :)
function writeDataToDb(product) {
    // the following code is here to create random delays and random failures (just for testing)
    var timeToWrite = Math.random() * 100;
    var failure = Math.random() > 0.5;
    return Promise.delay(timeToWrite).then(function() {
        if (failure) {
            throw new Error();
        }
        return null;
    })
}

xml.on('endElement: product', writeXmlDataWithBackPressure);

玩它，输入一些console.log 以了解它的行为方式。 我希望这能帮助你解决你的问题:)