4

我在一个项目中使用 Python 和celery 。在项目中,我有两个文件:

芹菜配置文件

BROKER_URL = "amqp://guest:guest@localhost:5672//"
CELERY_RESULT_BACKEND = "amqp"
CELERY_IMPORTS = ("example",)
CELERYD_CONCURRENCY = 2

example.py

from celery.task import task
import hashlib

md5 = hashlib.md5()

@task
def getDigest(text):
    print 'Using md5 - ',md5
    md5.update(text)
    return md5.digest()

celeryconfig.py中,我将CELERYD_CONCURRENCY设置为2,这意味着它将我的任务队列中的任务分配给2 个不同的进程。

从 Python 控制台,我运行:

from example import getDigest
getDigest.delay('foo');getDigest.delay('bar')

这将创建两个由两个工作人员同时执行的任务。问题是,由于两个工作进程都运行它们的任务函数 [ getDigest() ],它们似乎使用相同的哈希对象 ( md5 )。celeryd的输出证实了这一点,如下所示。

[PoolWorker-2] Using md5 -
[PoolWorker-2] <md5 HASH object @ 0x23e6870>
[PoolWorker-1] Using md5 -
[PoolWorker-1] <md5 HASH object @ 0x23e6870>

为了简单起见,我使用的是hashlib的md5对象,但是在我的实际项目中,我使用的是一个不能被多个进程访问和修改的对象。这预计会使工人崩溃。

这就提出了一个问题:如何修改我的代码以使工作进程初始化并使用它们自己的 ( md5 ) 对象?现在,他们共享同一个对象——导致我的应用程序崩溃。这可能吗?

4

1 回答 1

7

他们使用的是同一个对象,因为您在代码中明确告诉他们这样做。通过在任务范围之外创建对象并在任务中使用它,您可以让所有工作人员访问共享对象。这是一个并发问题,不一定是 Celery 问题。如果对象很小,您可以使用对象的副本,或者使用您自己的锁定策略。但是,一般来说,如果一个对象一次要被多个进程更新,它需要采用某种同步,这超出了 Celery 的范围。

于 2011-12-28T04:33:59.347 回答