一种确保在 MongoDb 的 findAndModify 中进行独占读取的方法? [英] A way to ensure exclusive reads in MongoDb's findAndModify?

问题描述

我有一个 MongoDB 集合(用作作业队列)，多个进程使用 findAndModify 从中读取记录.FindAndModify 搜索 active 字段为false"的记录，将其设置为true"，这样其他进程就不会读取相同的记录.

I have a MongoDB collection (used as a job queue), from which multiple processes read records, using findAndModify. FindAndModify searches for records where the active field is "false", setting it to "true", so that other processes do not read the same record.

问题是查看日志，我看到不同的进程仍然读取相同的记录.这似乎发生在两个进程同时从队列中读取时.有没有办法确保一个进程一次只从一个集合中读取数据?

The problem is that looking at logs, I see that different processes still read the same records. This seems to occur when two processes read from the queue at the same time. Is there any way to make sure that one only process reads from a collection at a time?

我使用的是 Mongo 2.2.3 和 pymongo 2.2.

I am using Mongo 2.2.3 and pymongo 2.2.

非常感谢！

有问题的日志是:

worker.3 2013-03-18 23:57:45,434 default-worker-3
project_name INFO 队列作业:作业 ID:5147a90f68e8fe0097002bdf

worker.3 2013-03-18 23:57:45,434 default-worker-3
project_name INFO Queue job: job id: 5147a90f68e8fe0097002bdf

worker.3 2013-03-18 23:57:47,608 default-worker-3
project_name INFO 输入:14497 个文档

worker.3 2013-03-18 23:57:47,608 default-worker-3
project_name INFO Input: 14497 docs

worker.2 2013-03-18 23:57:45,440 default-worker-2
project_name INFO 队列作业:作业 ID:5147a90f68e8fe0097002bdf

worker.2 2013-03-18 23:57:45,440 default-worker-2
project_name INFO Queue job: job id: 5147a90f68e8fe0097002bdf

worker.2 2013-03-18 23:57:47,658 default-worker-2
project_name INFO 输入:14497 个文档

worker.2 2013-03-18 23:57:47,658 default-worker-2
project_name INFO Input: 14497 docs

如您所见，worker.3 和 worker.2 从队列中读取相同的作业(两个 worker 的 mongodb id 相同).

As you can see, worker.3 and worker.2 read the same job from the queue (it has the same mongodb id for both workers).

find_and_modify 命令:

The find_and_modify command:

query = {"active": False}
try:
    return self.collection.find_and_modify(
            query=query,
            update={"$set": {"active": True}},
            upsert=False,
            sort={"added_on": 1},
            limit=1
        )
except Exception, exc:
    LOGGER.exception(exc)

推荐答案

先直接查询 mongodb 获取单个作业记录，

First make a straightforward query to mongodb to get a single job record,

job = db.coll.find({query}).limit(1)

接下来更新指定作业 id 和 where active=false

Next update the record specifying the job id and where active=false

update_response = db.coll.update(
    {_id:job.id, active=false},
    {$set:{active:true}},
    false,
    false
)

如果作业已经被其他进程更新，由于active=false的查询约束，更新不会成功.在 update_response 中检查记录已更新:

If the job has already been updated by another process, the update will not succeed due to the query constraint of active=false. Check in the update_response that the record was updated:

if update_response['n'] > 0 and update_response['updatedExisting']==true:
    return job

如果您的更新没有成功，请获取另一个作业并重试.

If your update didn't succeed then fetch another job and try again.

这篇关于一种确保在 MongoDb 的 findAndModify 中进行独占读取的方法?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！