如何按任务名称检查和取消Celery任务

Question

我正在使用Celery(3.0.15)和Redis作为经纪人.

是否有一种直接的方法来查询Celery队列中存在的具有给定名称的任务数量？

并且,作为后续,有没有办法取消Celery队列中存在的给定名称的所有任务？

我已经阅读了" 监控和管理指南",但没有看到解决方案.

Answer 1

# Retrieve tasks
# Reference: http://docs.celeryproject.org/en/latest/reference/celery.events.state.html
query = celery.events.state.tasks_by_type(your_task_name)

# Kill tasks
# Reference: http://docs.celeryproject.org/en/latest/userguide/workers.html#revoking-tasks
for uuid, task in query:
    celery.control.revoke(uuid, terminate=True)

Answer 2

有一个问题,早期的答案没有解决,如果他们不知道它可能会甩掉他们.

在已发布的解决方案中,我将Danielle用于一个小修改:我将任务导入到我的文件中并使用其.name属性来获取要传递给的任务名称.tasks_by_type().

app.control.revoke(
    [uuid for uuid, _ in
     celery.events.state.State().tasks_by_type(task.name)])

但是,此解决方案将忽略已安排用于将来执行的那些任务.像一些评论其他答案的人一样,当我查看.tasks_by_type()返回时我有一个空列表.事实上我的队列是空的.但我知道有些任务计划在未来执行,这些是我的主要目标.我可以通过执行来看到它们,celery -A [app] inspect scheduled但它们不受上面代码的影响.

我设法通过这样做来撤销计划任务:

app.control.revoke(
    [scheduled["request"]["id"] for scheduled in
     chain.from_iterable(app.control.inspect().scheduled()
                         .itervalues())])

app.control.inspect().scheduled()返回一个字典,其键是工作者名称,值是调度信息列表(因此,需要chain.from_iterable从中导入itertools).任务信息位于"request"调度信息的字段中,并"id"包含任务ID.请注意,即使在撤销后,计划任务仍将显示在计划任务中.已撤销的计划任务将不会从计划任务列表中删除,直到其计时器到期或Celery执行某些清理操作.(重新启动工作程序会触发此类清理.)

Answer 3

与 Celery 一样，这里的所有答案都不适合我，所以我做了我通常的事情，编写了一个直接检查 redis 的解决方案。开始了...

# First, get a list of tasks from redis:
import redis, json

r = redis.Redis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DATABASES['CELERY'],
)
l = r.lrange('celery', 0, -1)

# Now import the task you want so you can get its name
from my_django.tasks import my_task

# Now, import your celery app and iterate over all tasks 
# from redis and nuke the ones that have a matching name.
from my_django.celery_init import app
for task in l:
     task_headers = json.loads(task)['headers']
     task_name = task_headers["task"]
     if task_name == my_task.name:
         task_id = task_headers['id']
         print("Terminating: %s" % task_id)
         app.control.revoke(task_id, terminate=True)

请注意，以这种方式撤销可能不会撤销预取的任务，因此您可能不会立即看到结果。

此外，这个答案不支持优先任务。如果您想修改它来做到这一点，您将需要我的其他答案中的一些技巧来 hacks redis。