3

所以我有一个创建工作目录并在那里完成所有工作的任务。该任务从服务器 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？ 其他一些信号也有它们，我查看了这些信号的发布源，由于某种原因，这个信号没有提供 task_id。

提供的sender任务包含一个 trace_task 函数：{'__trace__': <function trace_task at 0x3ee8230>}但我看不到如何使用它，因为该函数本身需要一个 task_id。

欢迎任何其他想法。

4

1 回答 1

1

我认为这里发生的事情是您使用的是旧版本的 celery，它不支持第一个“请求”参数。

添加此内容的上游问题是 [1]；在此之前，我认为你运气不好，不幸的是没有办法获得 task_id。

[1] https://github.com/celery/celery/issues/1555

于 2013-11-14T12:51:10.600 回答