django.core.cache.lock 在 Celery 任务中不起作用

【问题标题】：django.core.cache.lock doesn't work in a Celery taskdjango.core.cache.lock 在 Celery 任务中不起作用
【发布时间】：2020-12-08 04:49:18
【问题描述】：

我有以下 (tasks.py)：

from celery import shared_task
from django.core.cache import cache

@shared_task
def test(param1: str) -> None:
    with cache.lock("lock-1"):
        print("hello")

当我执行test.delay() 时，没有打印任何内容，这让我相信cache.lock("lock-1") 有问题。

我使用 Redis 作为我的缓存和 Celery 后端，这是在 settings.py 中配置的。

这里有什么问题？如果django.core.cache 不能用作锁定机制（以确保一次只运行一个test，可以使用什么来代替？）谢谢！

【问题讨论】：

标签： python django celery

【解决方案1】：

找出原因 - 结果我没有阅读文档。

添加这个工作：

# https://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html
@contextmanager
def redis_lock(lock_id):
    timeout_at = time.monotonic() + LOCK_EXPIRE - 3
    # cache.add fails if the key already exists
    # Second value is arbitrary
    status = cache.add(lock_id, "lock", timeout=LOCK_EXPIRE)
    try:
        yield status
    finally:
        # memcache delete is very slow, but we have to use it to take
        # advantage of using add() for atomic locking
        if time.monotonic() < timeout_at and status:
            # don't release the lock if we exceeded the timeout
            # to lessen the chance of releasing an expired lock
            # owned by someone else
            # also don't release the lock if we didn't acquire it
            cache.delete(lock_id)

然后，lock语句就变成了

with redis_lock("lock-1"):

【讨论】：