zoukankan      html  css  js  c++  java
  • celery 与 flask 实现异步任务调度

        Flask 定了2中上下文,来实现机遇线程协程的,wsgi服务的请求(request、session)和存储(g,current_app )过程,通过栈来完成不同线程和协程的上下文切换,在与celery相结合处理异步任务时,需要保证异步任务在同一个上下文中执行,需要对celery进行重构, 避免出现:

    ile "/opt/xAssets/venv/lib/python3.6/site-packages/celery/app/trace.py", line 412, in trace_task
        R = retval = fun(*args, **kwargs)
      File "/opt/xAssets/venv/lib/python3.6/site-packages/celery_context/model.py", line 42, in __call__
        with current_app.app_context():
      File "/opt/xAssets/venv/lib/python3.6/site-packages/werkzeug/local.py", line 347, in __getattr__
        return getattr(self._get_current_object(), name)
      File "/opt/xAssets/venv/lib/python3.6/site-packages/werkzeug/local.py", line 306, in _get_current_object
        return self.__local()
      File "/opt/xAssets/venv/lib/python3.6/site-packages/flask/globals.py", line 51, in _find_app
        raise RuntimeError(_app_ctx_err_msg)
    RuntimeError: Working outside of application context.
    
    This typically means that you attempted to use functionality that needed
    to interface with the current application object in some way. To solve
    this, set up an application context with app.app_context().  See the
    documentation for more information.

    celery 的使用

    •  应用:
    from celery import Celery
    
    celery = Celery('xassets',
                    broker=settings.CELERY_BROKER_URL,
                    include="xassets.tasks",
                    backend=settings.CELERY_RESULT_BACKEND,
                    )
    •  celery 以 默认方式(perfork)
    # 紧接上文

      from celery.signals import worker_process_init

    TaskBase = celery.Task
    
    
    class ContextTask(TaskBase):
        abstract = True
    
        def __call__(self, *args, **kwargs):
            with current_app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    
    
    celery.Task = ContextTask
    
    
    @worker_process_init.connect
    def init_celery_flask_app(**kwargs):
        """Create the Flask app after forking a new worker.
    
        This is to make sure no resources are shared between processes.
        """
        app = create_app()
        app.app_context().push()

    # 注意: worker_process_init 信号,仅用于perfork方式启动,如果以gevent,greelet 或者thread方式启动,这此处不执行,也就是会缺少缺少app实例化

    •  以gevent方式启动(大多数人的选择)
    TaskBase = celery.Task
    
    app = create_app()
    
    
    class ContextTask(TaskBase):
        abstract = True
    
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    
    
    celery.Task = ContextTask

    # 注意:1. 需要提前创建 flask 上下文应用实例

                2. 重定义task时,建议使用app.app_context

    celery_context 封装一键化应用

    from celery_context import Celery
    
    celery = Celery('xassets',
                    broker=settings.CELERY_BROKER_URL,
                    include="xassets.tasks",
                    backend=settings.CELERY_RESULT_BACKEND,
                    )
    
    app = create_app()
    celery.reload_task(app)
    # 或者 
    # celery.init_app(app)
  • 相关阅读:
    Enter passphrase
    laravel扩展包-私有库
    php7中异常
    laravel-admin(自定义表单视图)
    laravel-admin(自定义表单与验证)
    ZJU校赛划水记
    bzoj 4762: 最小集合
    WC2017 颓废记
    bzoj 2337: [HNOI2011]XOR和路径
    bzoj 2870: 最长道路tree
  • 原文地址:https://www.cnblogs.com/spaceapp/p/14501918.html
Copyright © 2011-2022 走看看