findOne 和 updateOne 的 Mongo DB 并发问题

Question

我遇到了更新同一文档的并发请求的问题。我没有使用findAndModify()，因为我需要访问文档的当前状态来进行更新，我认为findAndModify() 不支持该更新。我也想避免使用db.fsyncLock()，因为它会锁定整个数据库，我只需要将一个文档锁定在一个集合中。

首先我使用findOne() 来获取一个文档，然后我在findOne() 的回调中使用updateOne() 来更新同一个文档。当我将一堆动作排队并同时运行它们时，我相信它们在调用 findOne() 时都在访问相同的状态，而不是等待 updateOne() 从上一个动作完成。

我应该如何处理？

mongoDBPromise.then((db)=> {
    db.collection("notes").findOne(
        {path: noteId},
        (err, result)=> {
            if (err) {
                console.log(err);
                return;
            }

            if (!result.UndoableNoteList.future.length) {
                console.log("Nothing to redo");
                return;
            }

            let past = result.UndoableNoteList.past.concat(Object.assign({},result.UndoableNoteList.present));
            let present = Object.assign({},result.UndoableNoteList.future[0]);
            let future = result.UndoableNoteList.future.slice(1, result.UndoableNoteList.future.length);

            db.collection("notes").updateOne(
                {path: noteId},
                {
                    $set: {
                        UndoableNoteList: {
                            past: past,
                            present: present,
                            future:future
                        }
                    }
                },
                (err, result)=> {
                    if (err) {
                        console.log(err);
                        return;
                    }
                }
            )
        }
    );
});

Answer 1

由于updateOne() 是一个异步调用，findOne() 不会等待它完成，因此可能会出现同一个文档同时更新的情况，这在 mongo 中是不允许的。

我认为updateOne() 在这种情况下是不必要的。 请注意，您已经在 findOne() 查询中找到了需要更新的文档的正确实例。现在，您无需执行updateOne() 即可更新该实例并保存该文档。我认为可以通过这种方式避免问题：

mongoDBPromise.then((db)=> {
    db.collection("notes").findOne(
        {path: noteId},
        (err, result)=> {
            if (err) {
                console.log(err);
                return;
            }

            if (!result.UndoableNoteList.future.length) {
                console.log("Nothing to redo");
                return;
            }

            let past = result.UndoableNoteList.past.concat(Object.assign({},result.UndoableNoteList.present));
            let present = Object.assign({},result.UndoableNoteList.future[0]);
            let future = result.UndoableNoteList.future.slice(1, result.UndoableNoteList.future.length);
            result.UndoableNoteList.past = past;
            result.UndoableNoteList.present = present;
            result.UndoableNoteList.future = future;

            //save the document here and return
        }
    );
});

希望这个答案对你有所帮助！

Answer 2

我无法找到使用纯 mongodb 函数顺序运行查询的方法。我编写了一些 node.js 逻辑来阻止 mongodb 查询在同一文档上运行并将这些查询添加到队列中。这是当前代码的样子。

Websocket 撤销监听器

module.exports = (noteId, wsHelper, noteWebSocket) => {
    wsHelper.addMessageListener((msg, ws)=> {
        if (msg.type === "UNDO") {
            noteWebSocket.broadcast(msg, noteWebSocket.getOtherClientsInPath(noteId, wsHelper));
            noteWebSocket.saveUndo(noteId);
        }
    });
};

从监听器调用的saveUndo函数

saveUndo(noteId) {
    this.addToActionQueue(noteId, {payload: noteId, type: "UNDO"});
    this.getNoteByIdAndProcessQueue(noteId);
}

从 saveUndo 调用的 getNoteByIdAndProcessQueue 函数

    getNoteByIdAndProcessQueue(noteId) {
        if (this.isProcessing[noteId])return;
        this.isProcessing[noteId] = true;
        mongoDBPromise.then((db)=> {
            db.collection("notes").findOne(
                {path: noteId},
                (err, result)=> {
                    if (err) {
                        this.isProcessing[noteId] = false;
                        this.getNoteByIdAndProcessQueue(noteId);
                        return;
                    }

                    this.processQueueForNoteId(noteId, result.UndoableNoteList);
                });
        });
    }

processQueueForNoteId 函数

processQueueForNoteId(noteId, UndoableNoteList) {

    this.actionQueue[noteId].forEach((action)=> {
        if (action.type === "UNDO") {
            UndoableNoteList = this.undoNoteAction(UndoableNoteList);
        } else if (action.type === "REDO") {
            UndoableNoteList = this.redoNoteAction(UndoableNoteList);
        } else if (action.type === "ADD_NOTE") {
            UndoableNoteList = this.addNoteAction(UndoableNoteList, action.payload);
        } else if (action.type === "REMOVE_NOTE") {
            UndoableNoteList = this.removeNoteAction(UndoableNoteList, action.payload);
        }
    });

    let actionsBeingSaved = this.actionQueue[noteId].concat();
    this.actionQueue[noteId] = [];
    mongoDBPromise.then((db)=> {
        db.collection("notes").updateOne(
            {path: noteId},
            {
                $set: {
                    UndoableNoteList: UndoableNoteList
                }
            },
            (err, result)=> {
                this.isProcessing[noteId] = false;

                // If the update failed then try again
                if (err) {
                    console.log("update error")
                    this.actionQueue[noteId] = actionsBeingSaved.concat(this.actionQueue[noteId]);
                }

                // if action were queued during save then save again
                if (this.actionQueue[noteId].length) {
                    this.getNoteByIdAndProcessQueue(noteId);
                }
            }
        )
    });
}