如何在 Celery 中检查任务状态？

Question

如何检查任务是否在 celery 中运行（具体来说，我正在使用 celery-django）？

我已经阅读了文档，并且用谷歌搜索了，但我看不到这样的调用：

my_example_task.state() == RUNNING

我的用例是我有一个用于转码的外部（java）服务。当我发送要转码的文档时，我想检查运行该服务的任务是否正在运行，如果没有，则（重新）启动它。

我相信我正在使用当前的稳定版本 - 2.4。

Answer 1

每个 Task 对象都有一个 .request 属性，其中包含 AsyncRequest 对象。因此，下面的行给出了任务 task 的状态：

task.AsyncResult(task.request.id).state

Answer 2

老问题，但我最近遇到了这个问题。

如果您尝试获取task_id，您可以这样做：

import celery
from celery_app import add
from celery import uuid

task_id = uuid()
result = add.apply_async((2, 2), task_id=task_id)

现在您确切地知道task_id是什么，并且现在可以使用它来获取AsyncResult：

# grab the AsyncResult 
result = celery.result.AsyncResult(task_id)

# print the task id
print result.task_id
09dad9cf-c9fa-4aee-933f-ff54dae39bdf

# print the AsyncResult's status
print result.status
SUCCESS

# print the result returned 
print result.result
4

Answer 3

您还可以创建自定义状态并在任务执行期间更新其值。
此示例来自文档：

@app.task(bind=True)
def upload_files(self, filenames):
    for i, file in enumerate(filenames):
        if not self.request.called_directly:
            self.update_state(state='PROGRESS',
                meta={'current': i, 'total': len(filenames)})

http://celery.readthedocs.org/en/latest/用户指南/tasks.html#custom-states

Answer 4

只需使用 celery FAQ 中的 API，

result = app.AsyncResult(task_id)

效果很好。

Answer 5

2020年答案：

#### tasks.py
@celery.task()
def mytask(arg1):
    print(arg1)

#### blueprint.py
@bp.route("/args/arg1=<arg1>")
def sleeper(arg1):
    process = mytask.apply_async(args=(arg1,)) #mytask.delay(arg1)
    state = process.state
    return f"Thanks for your patience, your job {process.task_id} \
             is being processed. Status {state}"

Answer 6

尝试：

task.AsyncResult(task.request.id).state

这将提供 Celery 任务状态。如果 Celery 任务已经处于 FAILURE 状态，它将抛出异常：

raised Outstanding: KeyError('exc_type',)

Answer 7

• 首先，在你的 celery APP 中：

vi my_celery_apps/app1.py

app = Celery(worker_name)

• 然后，切换到任务文件，从你的 celery 应用模块导入 app。

vi 任务/task1.py

from my_celery_apps.app1 import app

app.AsyncResult(taskid)

try:
   if task.state.lower() != "success":
        return
except:
    """ do something """

Answer 8

中找到了有用的信息

我在Celery 项目工作人员指南 Inspection-workers< /a>

对于我的情况，我正在检查 Celery 是否正在运行。

inspect_workers = task.app.control.inspect()
if inspect_workers.registered() is None:
    state = 'FAILURE'
else:
    state = str(task.state) 

您可以使用检查来满足您的需求。

Answer 9

res = method.delay()
    
print(f"id={res.id}, state={res.state}, status={res.status} ")

print(res.get())

Answer 10

对于简单的任务，我们可以使用 http://flower.readthedocs.io/en/latest /screenshots.html 和 http://policystat.github.io/jobtastic/到进行监控。

对于复杂的任务，比如涉及许多其他模块的任务。我们建议手动记录特定任务单元的进度和消息。

Answer 11

除了上述程序化方法之外
使用Flower可以轻松查看任务状态。

使用 Celery Events 进行实时监控。
Flower 是一个基于 Web 的工具，用于监视和管理 Celery 集群。

1. 任务进度和历史记录
2. 能够显示任务详细信息（参数、开始时间、运行时间等）
3. 图表和统计

官方文档：
Flower - Celery 监控工具

安装：

$ pip install flower

使用：

http://localhost:5555

更新：
这与版本控制有关，flower（版本=0.9.7）仅适用于celery（版本=4.4.7），而且当您安装flower时，它会将更高版本的celery卸载到4.4.7，这永远不适用于已注册的任务

Answer 12

返回task_id（从.delay()给出）并随后向celery实例询问状态：

x = method.delay(1,2)
print x.task_id

询问时，使用此task_id获取新的AsyncResult：

from celery.result import AsyncResult
res = AsyncResult("your-task-id")
res.ready()

Answer 13

从任务 ID 创建 AsyncResult 对象是 常见问题解答获取任务当你唯一拥有的就是任务时的状态 ID。

然而，从 Celery 3.x 开始，有一些重要的警告，如果不注意的话可能会咬人。这实际上取决于具体的用例场景。

默认情况下，Celery 不记录“运行”状态。

为了让 Celery 记录任务正在运行，您必须设置 task_track_started 为 True。下面是一个测试此功能的简单任务：

@app.task(bind=True)
def test(self):
    print self.AsyncResult(self.request.id).state

当选项 task_track_started 为 False（默认值）时，状态显示为 PENDING，即使任务已经开始。如果您将 task_track_started 设置为 True，则状态将为 STARTED。您可以覆盖特定任务的默认设置：

@app.task(bind=True, track_started=True)
def test(self):
    print self.AsyncResult(self.request.id).state

或者更改 settings.py 中的全局设置：

CELERY_TRACK_STARTED=True

状态 PENDING 表示“我不知道”。

状态为 PENDING 的 AsyncResult 仅仅意味着 Celery 不知道任务的状态。这可能是由于多种原因造成的。

一方面，AsyncResult 可以使用无效的任务 ID 来构造。这样的“任务”将被 Celery 视为待处理：

>>> task.AsyncResult("invalid").status
'PENDING'

好吧，所以没有人会向 AsyncResult 提供显然无效的 id。这很公平，但它还具有以下效果：AsyncResult 也会将已成功运行但 Celery 忘记为 PENDING 的任务视为PENDING。同样，在某些用例场景这可能是一个问题。部分问题取决于如何配置 Celery 来保存任务结果，因为它取决于结果后端中“墓碑”的可用性。 （“Tombstones”是 Celery 文档中用于记录任务如何结束的数据块的术语。）如果 task_ignore_result 是正确。更令人烦恼的问题是 Celery 默认会使墓碑过期。 result_expires默认设置为 24 小时。因此，如果您启动一个任务，并将 ID 记录在长期存储中，24 小时后，您用它创建一个 AsyncResult，状态将为 PENDING。

所有“实际任务”都以 PENDING 状态开始。因此，在任务上获得 PENDING 可能意味着该任务已被请求，但从未取得比这更进一步的进展（无论出于何种原因）。或者这可能意味着任务已运行，但 Celery 忘记了它的状态。

哎哟! AsyncResult 对我不起作用。我还能做什么？

我更喜欢跟踪目标而不是跟踪任务本身。我确实保留了一些任务信息，但这对于跟踪目标来说确实是次要的。目标存储在独立于 Celery 的存储中。当请求需要执行依赖于已实现的某个目标的计算时，它会检查该目标是否已经实现，如果是，则使用这个缓存的目标，否则，它启动会影响该目标的任务，并发送向发出 HTTP 请求的客户端发送一个响应，指示它应该等待结果。

---

上面的变量名称和超链接适用于 Celery 4.x。在 3.x 中，相应的变量和超链接为： CELERY_TRACK_STARTED，CELERY_IGNORE_RESULT，CELERY_TASK_RESULT_EXPIRES。