查看 celery 任务是否存在

Question

是否有可能找出具有特定任务ID的任务是否存在？当我尝试获取状态时，我将始终处于待处理状态。

>>> AsyncResult('...').status
'PENDING'

我想知道给定的任务 id 是否是真正的 celery 任务 id 而不是随机字符串。我想要不同的结果，具体取决于某个 id 是否存在有效任务。

过去可能存在具有相同 id 的有效任务，但结果可能已从后端删除。

Answer 1

Celery 在发送任务时不会写入状态，这部分是一种优化（参见documentation）。

如果你真的需要，添加很简单：

from celery import current_app
# `after_task_publish` is available in celery 3.1+
# for older versions use the deprecated `task_sent` signal
from celery.signals import after_task_publish

# when using celery versions older than 4.0, use body instead of headers

@after_task_publish.connect
def update_sent_state(sender=None, headers=None, **kwargs):
    # the task may not exist if sent using `send_task` which
    # sends tasks by name, so fall back to the default result backend
    # if that is the case.
    task = current_app.tasks.get(sender)
    backend = task.backend if task else current_app.backend
 
    backend.store_result(headers['id'], None, "SENT")

然后您可以测试 PENDING 状态以检测任务没有（看似） 已发送：

>>> result.state != "PENDING"

Answer 2

如果任务 ID 未知，AsyncResult.state 返回 PENDING。

待处理

任务正在等待执行或未知。任何不是的任务 ID known 暗示处于待处理状态。

http://docs.celeryproject.org/en/latest/userguide/tasks.html#pending

如果您需要区分未知 ID 和现有 ID，您可以提供自定义任务 ID：

>>> from tasks import add
>>> from celery.utils import uuid
>>> r = add.apply_async(args=[1, 2], task_id="celery-task-id-"+uuid())
>>> id = r.task_id
>>> id
'celery-task-id-b774c3f9-5280-4ebe-a770-14a6977090cd'
>>> if not "blubb".startswith("celery-task-id-"): print "Unknown task id"
... 
Unknown task id
>>> if not id.startswith("celery-task-id-"): print "Unknown task id"
... 

Answer 3

现在我正在使用以下方案：

1. 获取任务 ID。
2. 设置为 memcache 键，如 'task_%s' % task.id 消息 'Started'。
3. 将任务 ID 传递给客户端。
4. 现在我可以从客户端监控任务状态（从任务消息设置到内存缓存）。
5. 从准备就绪的任务 - 设置为内存缓存键消息“准备就绪”。
6. 从任务就绪的客户端开始 - 启动特殊任务，该任务将从内存缓存中删除密钥并执行必要的清理操作。

Answer 4

您需要在您创建的 AsyncTask 对象上调用 .get() 以实际从后端获取结果。

请参阅Celery FAQ。

---

进一步澄清我的回答。

任何字符串在技术上都是有效的 ID，没有办法验证任务 ID。找出任务是否存在的唯一方法是询问后端是否知道它，并且您必须使用.get()。

这引入了.get() 在后端没有关于您提供的任务 ID 的任何信息时阻塞的问题，这是设计为允许您启动任务然后等待其完成的问题。

在原始问题的情况下，我将假设 OP 想要获取先前完成的任务的状态。为此，您可以传递一个非常小的超时并捕获超时错误：

from celery.exceptions import TimeoutError
try:
    # fetch the result from the backend
    # your backend must be fast enough to return
    # results within 100ms (0.1 seconds)
    result = AsyncResult('blubb').get(timeout=0.1)
except TimeoutError:
    result = None

if result:
    print "Result exists; state=%s" % (result.state,)
else:
    print "Result does not exist"

不用说，这仅在您的后端存储结果时才有效，如果不是，则无法知道任务 ID 是否有效，因为没有记录它们。

---

进一步澄清。

使用 AMQP 后端无法完成您想做的事情，因为it does not store results, it forwards them。

我的建议是切换到数据库后端，以便结果在数据库中，您可以在现有 celery 模块之外查询。如果结果数据库中不存在任何任务，则可以假定 ID 无效。

Answer 5

所以我有这个想法：

import project.celery_tasks as tasks

def task_exist(task_id):
  found = False
  # tasks is my imported task module from celery
  # it is located under /project/project, where the settings.py file is located
  i = tasks.app.control.inspect()
  s = i.scheduled()
  for e in s:
    if task_id in s[e]:
      found = True
      break
  a = i.active()
  if not found:
    for e in a:
      if task_id in a[e]:
        found = True
        break
  r = i.reserved()
  if not found:
    for e in r:
      if task_id in r[e]:
        found = True
        break
  # if checking the status returns pending, yet we found it in any queues... it means it exists...
  # if it returns pending, yet we didn't find it on any of the queues... it doesn't exist
  return found

根据https://docs.celeryproject.org/en/stable/userguide/monitoring.html，不同类型的队列检查是： 积极的， 预定的， 预订的， 撤销， 挂号的， 统计数据， 查询任务，

请随意挑选。

而且可能有更好的方法来检查队列中的任务，但现在这对我来说应该有用。

Answer 6

试试

AsyncResult('blubb').state

这可能有效。

它应该返回不同的东西。

Answer 7

如果我错了，请纠正我。

if built_in_status_check(task_id) == 'pending'
   if registry_exists(task_id) == true
      print 'Pending'
   else
      print 'Task does not exist'