小编典典

芹菜periodic_task并行运行多次

python

我有一些使用Celery线程的非常简单的定期代码。它只是打印“ Pre”和“
Post”,然后在它们之间睡眠。此链接的网站改编而成

from celery.task import task
from celery.task import periodic_task
from django.core.cache import cache
from time import sleep
import main
import cutout_score
from threading import Lock

import socket
from datetime import timedelta
from celery.decorators import task, periodic_task

def single_instance_task(timeout):
  def task_exc(func):
    def wrapper(*args, **kwargs):
        lock_id = "celery-single-instance-" + func.__name__
        acquire_lock = lambda: cache.add(lock_id, "true", timeout)
        release_lock = lambda: cache.delete(lock_id)
        if acquire_lock():
            try:
                func()
            finally:
                release_lock()
    return wrapper
  return task_exc

LOCK_EXPIRE = 60 * 5 # Lock expires in 5 minutes
@periodic_task(run_every = timedelta(seconds=2))
def test():
    lock_id = "lock"

    # cache.add fails if if the key already exists
    acquire_lock = lambda: cache.add(lock_id, "true", LOCK_EXPIRE)
    # memcache delete is very slow, but we have to use it to take
    # advantage of using add() for atomic locking
    release_lock = lambda: cache.delete(lock_id)

    if acquire_lock():
        try:
            print 'pre'
            sleep(20)
            print 'post'
        finally:
            release_lock()
        return
    print 'already in use...'

此代码 永不 打印'already in use...'; 使用@single_instance_task装饰器时也会发生相同的现象。

你知道怎么了吗

编辑:我简化了问题,以便它不会写入内存(使用全局或django缓存); 我还是没看'already in use...'


编辑:当我将以下代码添加到我的Django settings.py文件中时(通过更改
https://docs.djangoproject.com/en/dev/topics/cache/中的代码,一切都按预期工作,
但仅当我使用port时11211 (奇怪的是,我的服务器在端口8000上)

CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.memcached.MemcachedCache',
        'LOCATION': [
            '127.0.0.1:11211'
        ]
    }
}

阅读 219

收藏
2021-01-20

共1个答案

小编典典

芹菜好吗 我对线程选项不熟悉。

如果它正在运行多进程,则没有在工作程序之间共享内存的“全局”变量。

如果您希望所有工作人员共享一个计数器,那么建议您使用cache.incr

例如:

In [1]: from django.core.cache import cache

In [2]: cache.set('counter',0)

In [3]: cache.incr('counter')
Out[3]: 1

In [4]: cache.incr('counter')
Out[4]: 2

更新资料

如果您通过休眠强制任务重叠,会发生什么情况,例如:

print "Task on %r started" % (self,)
sleep(20)
print "Task on %r stopped" % (self,)

如果您没有获得“已经在使用中…”的运行频率,则该时间超过20秒,那么您就会知道缓存的行为不符合预期。


另一个更新

您是否已在django设置中设置了缓存后端?例如memcached

如果不是,您可能正在使用 实际上不进行任何缓存Dummy
Cache
只需实现该接口即可 ……这听起来像是令人信服的问题原因。

2021-01-20