所以我有一个任务是创建一个工作目录并在那里完成它的所有工作。 该任务从服务器 A 调用并在工作服务器 上执行。
我需要确保在任务完成/取消后删除工作目录。
我添加了一个任务撤销处理程序，它看起来像这样:
@task
def my_task(value):
task_id = current_task.request.id
work_dir = os.path.join(BASE_WORK_DIR, task_id)
os.makedirs(work_dir)
try:
# Do work...
finally:
shutil.rmtree(work_dir)
@task_revoked.connect(sender=my_task)
def my_task_revoked_handler(*args, **kwargs):
# FIXME: delete work_dir
print args
# ()
print kwargs
# {'terminated': True, 'signal': <Signal: Signal>, 'expired': False, 'sender': <@task: myapp.core.tasks.my_task>, 'signum': '15'}
我的问题是当服务器 A 取消任务时，我无法在已撤销的处理程序中发出工作目录的清理，因为它没有 task_id。
有没有办法从这个特定的信号处理程序中获取任务 ID？ 一些other Signals有它们，我查看了发出这些信号的来源，由于某种原因，这个信号没有随 task_id 一起提供。
提供的sender task 包含一个 trace_task 函数:{'__trace__': <function trace_task at 0x3ee8230>}但我看不出如何使用它，因为该函数本身需要一个 task_id。
欢迎任何其他想法。
最佳答案
我认为这里发生的事情是您使用的是旧版本的 celery ，它不支持第一个“请求”参数。
添加这个的上游问题是[1]；在此之前，我认为你运气不好，不幸的是没有办法获得 task_id。
关于python - task_revoked 处理程序中的 Celery 任务 ID，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/15633073/