9

通过继承 PeriodicTask，我设法让周期性任务在 django-celery 中工作。我试图创建一个测试任务并让它运行一些无用的东西。有用。

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

我已经看到我应该“撤销”它的建议，但为此我似乎需要一个任务 ID，而且我看不到如何找到任务 ID。

谢谢。

4

2 回答 2

21

A task is a message, and a "periodic task" sends task messages at periodic intervals. Each of the tasks sent will have an unique id assigned to it.

revoke will only cancel a single task message. To get the id for a task you have to keep track of the id sent, but you can also specify a custom id when you send a task.

I'm not sure if you want to cancel a single task message, or if you want to stop the periodic task from sending more messages, so I'll list answers for both.

There is no built-in way to keep the id of a task sent with periodic tasks, but you could set the id for each task to the name of the periodic task, that way the id will refer to any task sent with the periodic task (usually the last one). You can specify a custom id this way,

either with the @periodic_task decorator:

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

or with the CELERYBEAT_SCHEDULE setting:

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

If you want to remove a periodic task you simply remove the @periodic_task from the codebase, or remove the entry from CELERYBEAT_SCHEDULE. If you are using the Django database scheduler you have to remove the periodic task from the Django Admin interface.

PS1: revoke doesn't stop a task that has already been started. It only cancels tasks that haven't been started yet. You can terminate a running task using revoke(task_id, terminate=True). By default this will send the TERM signal to the process, if you want to send another signal (e.g. KILL) use revoke(task_id, terminate=True, signal="KILL").

PS2: revoke is a remote control command so it is only supported by the RabbitMQ and Redis broker transports. If you want your task to support cancellation you should do so by storing a cancelled flag in a database and have the task check that flag when it starts:

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)
于 2011-11-22T18:23:07.730 回答
6

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

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

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

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

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

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

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

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

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

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

我希望这有帮助！

于 2015-10-09T21:56:10.567 回答