我不了解具体的Mongodb驱动程序和这个条目缓冲区。也许它只在特定情况下保留数据。
因此,我将用更通用的方法回答这个问题,适用于任何数据库。
总之,您有两个问题:
1. 您无法从失败的尝试中恢复
2. XML流发送数据过快
为了处理第一个问题,您需要实现一个重试算法,以确保在放弃之前进行多次尝试。
为了处理第二个问题,您需要在xml流上实施反压。您可以使用“暂停”方法、“恢复”方法和输入缓冲区来实现。
var Promise = require('bluebird');
var fs = require('fs');
var Xml = require('xml-stream');
var fileStream = fs.createReadStream('myFile.xml');
var xml = new Xml(fileStream);
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 = [];
function writeXmlDataWithBackPressure(product) {
var tryStartTask = function() {
if (!stopped && !suspended && currentPressure >= maxPressure) {
xml.pause();
suspended = true;
console.log("stream paused");
}
if (currentPressure < maxPressure) {
if (buffer.length > 0) {
buffer.shift()();
} else if (!stopped) {
try {
xml.resume();
suspended = false;
console.log("stream resumed");
} catch (e) {
stopped = true;
console.log("stream end");
}
}
}
};
buffer.push(function() {
currentPressure++;
exponentialRetry(function() {
return writeDataToDb(product)
}, 100, 2000, 100).finally(function() {
currentPressure--;
tryStartTask();
});
});
tryStartTask();
}
function writeDataToDb(product) {
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
来了解它的行为。希望这能帮助您解决问题 :)
xml-stream
进行的测试都会在Mongo服务器恢复后插入缓冲区对象。也许您可以发布更多代码/提供有关设置的更多信息? - cviejo