标签:地方 queue 申请 默认 完成 send broker class end
记一次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哦!
标签:地方 queue 申请 默认 完成 send broker class end
原文地址:https://blog.51cto.com/14612701/2544203