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)

posted @ 2021-03-08 20:18 binger0712 閱讀(475) 評論(2) 收藏舉報

刷新頁面返回頂部

binger0712

celery 與 flask 實現異步任務調度

celery 的使用

celery_context 封裝一鍵化應用

公告