关闭 Celery 任务

Question

我正在尝试找到一种能够从 django 管理员打开/关闭 celery 任务的方法。这主要是为了在服务关闭或有计划的维护期时禁用调用外部服务的任务。

对于我的定期任务，这很容易，尤其是使用 django-celery。但是对于按需调用的任务，我遇到了一些麻烦。目前，我正在探索仅在 TaskControl 模型中存储各种任务的开/关状态，然后在任务执行开始时检查该状态，如果状态为 False，则返回 None。由于每次任务开始时都会进行所有额外的数据库查找，这让我感到很脏。我可以使用不是 db 的缓存后端，但是为这些少数键/值对添加缓存似乎有点矫枉过正。

在models.py中

# this is a singleton model. singleton code bits omitted for brevity.
class TaskControl(models.Model):
    some_status = models.BooleanField(default=True)
    # more statuses

在tasks.py中

@celery.task(ignore_result=True)
def some_task():
    task_control = TaskControl.objects.get(pk=1)
    if not task_control.some_status:
        return None

    # otherwise execute task as normal

有什么更好的方法来做到这一点？

Answer 1

选项 1。尝试您的简单方法。看看会不会影响性能。如果没有，就失去“肮脏”的感觉。

选项 2. 使用单例缓存在进程内存中。为您的TaskControl 模型添加新鲜度信息：

class TaskControl(models.Model):
    some_status = models.BooleanField(default=True)
    # more statuses
    expires = models.DateTimeField()
    check_interval = models.IntegerField(default=5 * 60)

    def is_stale(self):
        return (
            (datetime.utcnow() >= self.expires) or
            ((datetime.utcnow() - self.retrieved).total_seconds >= self.check_interval))

然后在task_ctl.py:

_control = None

def is_enabled():
    global _control
    if (_control is None) or _control.is_stale():
        _control = TaskControl.objects.get(pk=1)
        # There's probably a better way to set `retrieved`,
        # maybe with a signal or a `Model` override,
        # but this should work.
        _control.retrieved = datetime.utcnow()
    return _control.some_status

选项 3。与选项 2 类似，但不是基于时间的过期，而是使用 Celery 的 remote control 强制所有工作人员重新加载 TaskControl（您必须编写自己的控制命令，而我没有知道您需要的所有内部组件是否都是公共 API）。

选项 4，仅适用于所有 Celery 工作人员在单台机器上运行的情况。将开/关标志作为文件存储在该机器的文件系统上。使用os.path.exists 查询它的存在（应该是单个stat() 或其他东西，足够便宜）。如果工作人员和管理面板在不同的机器上，请使用特殊的 Celery 任务来创建/删除文件。

选项 5。与选项 4 类似，但在 admin/web 机器上：如果文件存在，则首先不要将任务排队。