芹菜-获取当前任务的任务 ID

如何从任务内部获取任务的 task _ id 值:

from celery.decorators import task
from django.core.cache import cache


@task
def do_job(path):
"Performs an operation on a file"


# ... Code to perform the operation ...


cache.set(current_task_id, operation_results)

其思想是，在创建任务的新实例时，从任务对象中检索 task_id。然后使用任务 ID 确定任务是否已完成。我 不要希望通过 path值跟踪任务，因为文件在任务完成后被“清理”，可能存在，也可能不存在。

在上面的例子中，如何得到 current_task_id的值？

64074 次浏览
小开
最佳答案

Celery does set some default keyword arguments if the task accepts them. (you can accept them by either using **kwargs, or list them specifically)

@task
def do_job(path, task_id=None):
cache.set(task_id, operation_results)

The list of default keyword arguments is documented here: http://ask.github.com/celery/userguide/tasks.html#default-keyword-arguments

小开

UPDATE: use Balthazar's answer for Celery 3.1+

@task(bind=True)
def do_job(self, path):
cache.set(self.request.id, operation_results)

Feel free to upvote his answer.

Old answer:

Since Celery 2.2.0, information related to the currently executed task is saved to task.request (it's called «the context»). So you should get task id from this context (not from keyword arguments, which are deprecated):

@task
def do_job(path):
cache.set(do_job.request.id, operation_results)

The list of all available fields is documented here: http://celery.readthedocs.org/en/latest/userguide/tasks.html?highlight=requestcontext#context

小开

As of celery 3.1, you can use the bind decorator argument, and have access to the current request:

@task(bind=True)
def do_job(self, path):
cache.set(self.request.id, operation_results)

提交答案