记一次celery使用过成中丢失上下文问题

Posted 2020-12-20

记一次celery使用过程中丢失上下文问题

问题场景

项目中需要在celery的异步任务中连接其他服务的celery任务队列，并向所连接的celery中发送task任务。
这样使用的结果就是：需要生成两个不同的celery实例，在第一个celery实例中需要使用flask中的上下文，并进行上下文的生成，在调用完另一个celery实例后， 在worker运行了标记的task最大任务销毁重启之后；会发生在需要上下文的celery异步任务中报没有flask上下文的错误。例如如下逻辑

from project import create_app()     #  封装的flask的创建app过程
from celery import Celery
from project.task import celery_config

celery_app = Celery(__name__, broker=celery_config.CELERY_BROKER_URL)   # 程序中默认celery_app
celery_app.config_from_object(‘project.task.celery_config‘)

@celery_app.task(name=‘example_task‘)
def example_task(**kwargs):
    from project import create_app
    app = create_app()
    app.app_context().push()

    # 执行默认celery异步任务

        # 执行完成后进行向其他服务发送celery task
    send_celery_task(**kwargs)

def send_celery_task(**kwargs):
    # 产生问题的位置
    push_celery_app = Celery(__name__, broker=os.environ.get("PUSH_CELERY_BROKER_URL"))

    push_celery_app.send_task(name=os.environ.get("PUSH_TASK_NAME",
                                                                queue=os.environ.get("PUSH_QUEUE"),
                                                                 kwargs=kwargs)

经过查询celery源码，发现当需要申请新的celery实例的时候，需要将set_as_current 设置为False， 默认为True，为True的时候会重新刷新上下文，但是此时申请的celery实例并没有flask的上下文，所以刷新的时候将会走默认值None，在执行完重建work的时候，会拿当前的celery中的上下文，但是当前为None，所以在之后需要上下文的任务执行的时候，就会产生没有flask上下文的错误。

修改方式

将产生问题地方的代码修改为：

 push_celery_app = Celery(__name__, broker=os.environ.get("PUSH_CELERY_BROKER_URL"),  set_as_current=False)

---

From: xaohuihui
手搓不易，还请star哦！