停止/清除 Django-Celery 中的定期任务

Question

我已经通过子类化PeriodicTask 设法让定期任务在django-celery 中工作。我尝试创建一个测试任务并将其设置为执行一些无用的操作。有用。

现在我无法阻止它。我已阅读文档，但无法找到如何从执行队列中删除任务。我尝试过使用 celeryctl 和 shell，但registry.tasks() 是空的，所以我不知道如何删除它。

我看到有人建议我应该“撤销”它，但为此我似乎需要一个任务 id，而且我不知道如何找到该任务 id。

谢谢。

Answer 1

以防万一这可能对某人有帮助......我们在工作中遇到了同样的问题，尽管我们努力寻找某种管理命令来删除定期任务，但我们做不到。所以这里有一些提示。

您可能应该首先仔细检查哪个 调度程序您正在使用的类。

默认调度程序是celery.beat.PersistentScheduler，它只是在本地数据库文件（搁置）中跟踪上次运行时间。

在我们的例子中，我们使用 djcelery。 Schedulers.DatabaseScheduler 类。

django-celery 还附带一个调度程序，用于将调度存储在 Django 数据库中

尽管文档确实提到了一种删除定期任务的方法：

使用 django-celery 的调度程序，您可以在 Django 管理中添加、修改和删除定期任务。

我们希望以编程方式或通过 shell 中的（celery/管理）命令来执行删除。

由于我们找不到命令行，我们使用了 django/python shell：

$ python manage.py shell
>>> from djcelery.models import PeriodicTask
>>> pt = PeriodicTask.objects.get(name='the_task_name')
>>> pt.delete()

我希望这有帮助！

Answer 2

任务就是消息，“周期性任务”以周期性的间隔发送任务消息。每个发送的任务都会分配一个唯一的 ID。

revoke 只会取消单个任务消息。要获取您必须保留的任务的 ID
跟踪发送的 ID，但您也可以在发送任务时指定自定义 ID。

我不确定您是否想取消单个任务消息，或者您是否想阻止定期任务发送更多消息，因此我将列出两者的答案。

没有内置方法来保存与定期任务一起发送的任务的 ID，
但是你可以将每个任务的 id 设置为定期任务的名称，这样
id 将引用与定期任务一起发送的任何任务（通常是最后一个任务）。
您可以通过这种方式指定自定义 ID，

可以使用 @periodic_task 装饰器：

@periodic_task(options={"task_id": "my_periodic_task"})
def my_periodic_task():
    pass

或使用 CELERYBEAT_SCHEDULE 设置：

CELERYBEAT_SCHEDULE = {name: {"task": task_name,
                              "options": {"task_id": name}}}

如果您想删除定期任务，只需删除 >@periodic_task 从代码库中删除，或从 CELERYBEAT_SCHEDULE 中删除该条目。
如果您使用 Django 数据库调度程序，则必须删除定期任务
从 Django 管理界面。

PS1：revoke 不会停止已经启动的任务。只会取消
尚未开始的任务。您可以使用以下命令终止正在运行的任务
撤销（task_id，终止=True）。默认情况下，这会将 TERM 信号发送到
进程，如果你想发送另一个信号（例如KILL）使用
撤销（task_id，终止=True，信号=“KILL”）。

PS2：revoke是远程控制命令，因此仅RabbitMQ支持
和 Redis 代理传输。
如果您希望任务支持取消，您应该通过存储 cancelled 来实现
数据库中的标志，并让任务在启动时检查该标志：

from celery.task import Task

class RevokeableTask(Task):
    """Task that can be revoked.

    Example usage:

        @task(base=RevokeableTask)
        def mytask():
            pass
    """

    def __call__(self, *args, **kwargs):
        if revoke_flag_set_in_db_for(self.request.id):
            return
        super(RevokeableTask, self).__call__(*args, **kwargs)