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

Question

我设法通过继承PeriodicTask来定期在django-celery中工作.我试图创建一个测试任务并将其设置为运行无用的东西.有用.

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

我已经看到了我应该"撤销"它的建议,但为此我似乎需要一个任务ID,我无法看到我将如何找到任务ID.

谢谢.

Answer 1

任务是消息,"周期性任务"以定期间隔发送任务消息.发送的每个任务都将分配一个唯一的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 Admin界面中删除定期任务.

PS1:revoke不会停止已经启动的任务.它只取消尚未启动的任务.您可以使用终止正在运行的任务 revoke(task_id, terminate=True).默认情况下TERM,如果要发送另一个信号(例如KILL),则会将信号发送到进程 revoke(task_id, terminate=True, signal="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)

Answer 2

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

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

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

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

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

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

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

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

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

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

我希望这有帮助！