编辑 6/15我尝试运行相同的代码，在调用之前添加延迟$doSafePush()再次收到 ConcurrencyDBError 时（即执行return when.resolve("wait").delay(35).then(function() {$doSafePush(err.object, "set", value, retries-1);});代替return $doSafePush(err.object, "set", value, retries-1);）。只要我等待至少 35 毫秒，它就可以正常工作；低于该阈值它可以工作有时。感觉像是一个神奇的数字，但它正在起作用......
如果有人有任何线索，仍然有兴趣了解为什么:)
在 Node.js 下对 MongoDB 中的单个文档进行并发读写操作时遇到问题
请参阅下面我正在运行的代码。
基本上我将一个 Collection 定义为 Mongodb 结构{_id: ObjectID, __occ: Number, set: Array}。由于我有多个事务要“并行”执行，因此当我仅在正在使用的版本具有正确的 _id 和正确的 __occ 时才更新文档时，我想使用经典的乐观一致性控制框架编写安全推送；否则我强制重新加载并重试。
本质上，它会导致在同一个文档上同时运行多个写入和读取（如果写入失败）操作。
不幸的是，我无法让它工作。在某些时候，Mongo 开始返回有趣的文档，这让一切都变得混乱。
从下面运行代码，我有以下输出（对应于下面最后一个函数最底部的 debug(...) 命令）
第 1 部分：工作正常
29033b successfully added >>> occ=1 set={#1 elts, ["29033b"]} +87ms
29033c successfully added >>> occ=2 set={#2 elts, ["29033b","29033c"]} +33ms
29033d successfully added >>> occ=3 set={#3 elts, ["29033b","29033c","29033d"]} +37ms
第 2 部分：第一次事故：前第 3 项被另一个项替换 => 第一个数据丢失
290342 successfully added >>> occ=3 set={#3 elts, ["29033b","29033c",**"290342"**]} +37ms
第 3 部分：第二次事故；回到原来的项目 #3 => '290342' 丢失
29033e successfully added >>> occ=4 set={#4 elts, ["29033b","29033c",**"29033d"**,"29033e"]} +32ms
290344 successfully added >>> occ=5 set={#5 elts, ["29033b","29033c","29033d","29033e","290344"]} +10ms
290343 successfully added >>> occ=6 set={#6 elts, ["29033b","29033c","29033d","29033e","290344","290343"]} +35ms
第 4 部分：第三次事故；我们回到第三位的项目“290342”；它变得丑陋了
29033f successfully added >>> occ=4 set={#4 elts, ["29033b","29033c",**"290342"**,"29033f"]} +31ms
第5部分：第四次事故；再次，项目 #3 发生了变化，并且第 3 部分中的项目 #4 丢失了
290340 successfully added >>> occ=5 set={#5 elts, ["29033b","29033c","29033d",**"29033e"**,**"290340"**]} +35ms
第 6 部分：第五次事故；第 4 项已更改
290341 successfully added >>> occ=5 set={#5 elts, ["29033b","29033c","29033d","29033e",**"290341"**]} +45ms
这是我在 MongoDB 中的内容（通过 mongo shell 查询）
{ "_id" : ObjectId("575f8981762120ce1a290324"), "__occ" : 5, "set" : [ "29033b", "29033c", "29033d", "29033e", "290340" ] }
我原以为 findOneAndUpdate 是原子的，但它的行为对我来说很奇怪。任何人都可以帮助我找出我做错了什么
提前致谢
PS：为了便于阅读，所有数字都被缩短，但它们实际上是正确的 MongoDB ID，例如575f8981762120ce1a29033b => 29033b
// ------ MAIN CODE ------
var document = 'a previously loaded document from MongoDB'; // has _id and __occ properties
var valueList = [
"29033b",
"29033c",
"29033d",
"29033e",
"29033f",
"290340",
"290341",
"290342",
"290343",
"290344"
];
var promiseArray = [];
for (let i = 0; i < valueList.length; i++) {
promiseArray.push(safePush(document, valueList[i]));
}
return when.all(promiseArray)
.then(function(array) {
// do something with results
})
// ------ FUNCTIONS USED ------
function safePush(obj, value) {
return $doSafePush(obj, "set", value)
.then(function(updatedObject) {
return updatedObject;
}).catch(function(err) {
if (err instanceof ConcurrencyDBError) {
if (retries > 0) {
return $doSafePush(err.object, "set", value, retries-1);
} else {
err.message += " => max number of retries reached, failing";
}
}
// else
var message = err.message || "ERROR Unexpected error";
return when.reject(GenericError.create(message, err));
});
}
function $doSafePush(obj, property, value) {
var $query = {
'_id': obj.id,
'__occ': obj.__occ
}
var $update = {
"$push": {}
"$inc": {
"__occ": 1
}
};
$update["$push"][property] = value; // Update == $push value into 'property' array and $inc __occ counter
return $getMongoCollection("Users").findOneAndUpdate($query, $update, {returnOriginal: false})
.then(function(r) {
var result = r.value;
if (result == null) {
// No object matching $query found => OCC error
// Load new object and reject Promise
return $getMongoCollection().findOne({_id: new ObjectID(obj._id)})
.then(function(newObj) {
return when.reject(new ConcurrencyDBError(util.format("ERROR Concurrency error trying to add %s to %s", value, property), newObj));
});
}
// Else Object was found and updated => return it
debug("BGCHECK 3 >>> OK >>> %s successfully added >>> occ=%d set={#%d elts, %j}", value, result.__occ, u.getProperty(result, property).length, u.getProperty(result, property));
return when.resolve(result);
});
}
