Django Celery shared_task 单例模式答案

【问题标题】：Django Celery shared_task singleton patternDjango Celery shared_task 单例模式
【发布时间】：2021-03-15 00:01:08
【问题描述】：

我有一个基于 Django 的站点，它有几个在 Celery 工作者中执行的后台进程。我有一个特定任务可以运行几秒钟，如果第二个任务尝试访问相同的行，则对数据库进行多次读/写操作会受到竞争条件的影响。

我试图通过确保该任务一次只在一个工作人员上运行来防止这种情况发生，但我遇到了让其正常工作的问题。我以Celery Task Cookbook Recipe 为灵感，尝试制作我自己的版本，以确保该特定任务一次仅在一个工作人员上运行，但似乎仍然有可能遇到它的情况跨多个工作人员执行。

到目前为止，在tasks.py 我有：

class LockedTaskInProgress(Exception):
    """The locked task is already in progress"""
    silent_variable_failure = True


@shared_task(autoretry_for=[LockedTaskInProgress], default_retry_delay=30)
def create_or_update_payments(things=None):
    """
    This takes a list of `things` that we want to process payments on. It will
    read the thing's status, then apply calculations to make one or more payments
    for various users that are owed money for the thing.

    `things` - The list of things we need to process payments on.
    """
    lock = cache.get('create_or_update_payments')  # Using Redis as our cache backend

    if not lock:
        logger.debug('Starting create/update payments processing. Locking task.')
        cache.set('create_or_update_payments', 'LOCKED')
        real_create_or_update_payments(things)  # Long running function w/ lots of DB read/writes
        cache.delete('create_or_update_payments')
        logger.debug('Completed create/update payments processing. Lock removed.')
    else:
        logger.debug('Unable to process create/update payments at this time. Lock detected.')
        raise LockedTaskInProgress

以上似乎几乎可以工作，但在我的测试中，cache.get 和 cache.set 之间似乎仍然存在可能的竞争条件。

我很想获得有关如何改进它以使其更健壮的建议。

【问题讨论】：

标签： django celery

【解决方案1】：

我想我已经找到了一种方法，灵感来自我之前使用的旧版 Celery Task Cookbook 食谱。

这是我的实现：

class LockedTaskInProgress(Exception):
    """The locked task is already in progress"""
    silent_variable_failure = True


@shared_task(autoretry_for=[LockedTaskInProgress], default_retry_delay=30)
def create_or_update_payments(things=None):
    """
    This takes a list of `things` that we want to process payments on. It will
    read the thing's status, then apply calculations to make one or more payments
    for various users that are owed money for the thing.

    `things` - The list of things we need to process payments on.
    """
    LOCK_EXPIRE = 60 * 5  # 5 Mins
    lock_id = 'create_or_update_payments'

    acquire_lock = lambda: cache.add(lock_id, 'LOCKED', LOCK_EXPIRE)
    release_lock = lambda: cache.delete(lock_id)

    if acquire_lock():
        try:
            logger.debug('Starting create/update payments processing. Locking task.')
            real_create_or_update_payments(things)  # Long running function w/ lots of DB read/writes
        finally:
            release_lock()
            logger.debug('Completed create/update payments processing. Lock removed.')
    else:
        logger.debug('Unable to process create/update payments at this time. Lock detected.')
        raise LockedTaskInProgress

很有可能有更好的方法来做到这一点，但这似乎在我的测试中有效。

【讨论】：