在Celery任务之间共享一个通用实用程序功能

Question

我在Celery中有很多任务都使用了canvas chain.

@shared_task(bind=True)
def my_task_A(self):
    try:
        logger.debug('running task A')
        do something
    except Exception:
        run common cleanup function

@shared_task(bind=True)
def my_task_B(self):
    try:
        logger.debug('running task B')
        do something else
    except Exception:
        run common cleanup function

...

到现在为止还挺好.问题是我正在寻找使用这样的通用实用程序函数的最佳实践:

def cleanup_and_notify_user(task_data):
    logger.debug('task failed')
    send email
    delete folders
    ...

没有任务阻止,最好的办法是什么？例如,我可以run common cleanup function用一个电话替换cleanup_and_notify_user(task_data)吗？如果来自多个工作人员的多个任务试图同时调用该函数会发生什么？

每个工人都有自己的副本吗？我显然对这里的一些概念感到困惑.任何帮助深表感谢.

谢谢大家.

Answer 1

从 celery 任务内部，您只需编写 python 代码，因此该任务有自己的进程，并且该函数将像任何基本 OOP 逻辑一样为每个任务实例化。当然，如果此清理功能尝试删除系统文件夹或数据库行等共享资源，您最终会遇到需要以其他方式解决的外部资源的并发访问问题，例如，在 fylesystem 的情况下，您可以为每个任务创建一个沙箱。希望这可以帮助