如何检查一项任务是否在 celery 中运行（具体来说，我使用的是 celery-django）？
我已经阅读了文档，并在 Google 上搜索过，但我看不到类似以下的调用：
my_example_task.state() == RUNNING
我的用例是我有一个用于转码的外部 (java) 服务。当我发送要转码的文档时，我想检查运行该服务的任务是否正在运行，如果没有，则（重新）启动它。
我正在使用当前的稳定版本 - 2.4，我相信。
【问题讨论】：
标签： python web-services celery django-celery
res = method.delay()
print(f"id={res.id}, state={res.state}, status={res.status} ")
print(res.get())
【讨论】：
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}"
【讨论】：
vi my_celery_apps/app1.py
app = Celery(worker_name)
vi 任务/task1.py
from my_celery_apps.app1 import app
app.AsyncResult(taskid)
try:
if task.state.lower() != "success":
return
except:
""" do something """
【讨论】：
【讨论】：
除了上述程序化方法 使用Flower Task状态一目了然。
使用 Celery Events 进行实时监控。 Flower 是一个基于 Web 的工具，用于监控和管理 Celery 集群。
官方文档： Flower - Celery monitoring tool
安装：
$ pip install flower
用法：
http://localhost:5555
【讨论】：
从任务 id 创建一个AsyncResult 对象是在FAQ 中推荐的方式来获取任务状态，当你唯一拥有的是任务 id 时。
但是，从 Celery 3.x 开始，如果人们不注意它们，有一些重要的警告可能会咬人。这真的取决于具体的用例场景。
为了让 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。
PENDING 状态表示“我不知道”。状态为PENDING 的AsyncResult 并不意味着Celery 不知道任务的状态。这可能是由于多种原因。
一方面，AsyncResult 可以使用无效的任务 ID 构造。这样的“任务”将被 Celery 视为未决：
>>> task.AsyncResult("invalid").status
'PENDING'
好的，所以没有人会将显然无效的 id 提供给AsyncResult。很公平，但它也有效果，AsyncResult 还将考虑一个已成功运行但 Celery 忘记为 PENDING 的任务。再次，在某些用例中场景 这可能是个问题。部分问题取决于如何配置 Celery 以保留任务结果，因为它取决于结果后端中“墓碑”的可用性。 （“墓碑”是 Celery 文档中用于记录任务如何结束的数据块的术语。）如果 task_ignore_result 是 True，则使用 AsyncResult 根本不起作用。一个更令人烦恼的问题是 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。
【讨论】：
我在
中找到了有用的信息Celery Project Workers Guide inspecting-workers
就我而言，我正在检查 Celery 是否正在运行。
inspect_workers = task.app.control.inspect()
if inspect_workers.registered() is None:
state = 'FAILURE'
else:
state = str(task.state)
您可以使用检查来满足您的需求。
【讨论】：
对于简单的任务，我们可以使用http://flower.readthedocs.io/en/latest/screenshots.html 和http://policystat.github.io/jobtastic/ 来做监控。
对于复杂的任务，比如说一个处理很多其他模块的任务。我们建议手动记录特定任务单元的进度和消息。
【讨论】：
试试：
task.AsyncResult(task.request.id).state
这将提供 Celery 任务状态。如果 Celery Task 已经处于 FAILURE 状态，它将抛出异常：
raised unexpected: KeyError('exc_type',)
【讨论】：
老问题，但我最近遇到了这个问题。
如果你想获取 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
【讨论】：
apply_async。 apply_async 返回的对象是一个AsyncResult 对象，它确实有 Celery 生成的任务的 id。
task_id 都不需要您自己生成 任务ID。在您的评论中，您想象了一个超越“我如何检查任务状态”和“如果您尝试获取 task_id...”的原因需要，但这里不是这样。（此外，使用 uuid() 生成任务 ID 绝对没有超出 Celery 默认所做的。）
您还可以创建自定义状态并在任务执行期间更新其值。 此示例来自文档：
@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/userguide/tasks.html#custom-states
【讨论】：
每个Task 对象都有一个.request 属性，其中包含AsyncRequest 对象。因此，以下行给出了任务task 的状态：
task.AsyncResult(task.request.id).state
【讨论】：
state 属性，它仍然返回PENDING。这似乎不是从终端跟踪芹菜任务状态的可靠方法。此外，我正在运行 Celery Flower（Celery 监控工具），由于某种原因，它没有在它已执行的任务列表中显示我正在寻找的任务。我可能需要查看 Flower 设置，看看是否有任何内容显示仅在过去的特定小时内显示。
返回 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()
【讨论】：
x 怎么办？
async_result 的实例化。在您的用例中，您已经拥有实例，您可以开始了。但是，如果您只有任务 ID，并且需要实例化一个 async_result 实例才能调用 async_result.get()，会发生什么？这是AsyncResult类的一个实例，但是你不能使用原始类celery.result.AsyncResult，你需要从app.task()包装的函数中获取类。在你的情况下，你会做async_result = run_instance.AsyncResult('task-id')
but you cannot use the raw class celery.result.AsyncResult, you need to get the class from the function wrapped by app.task(). - 我认为这就是它的实际使用方式。阅读代码：github.com/celery/celery/blob/…