Flask后端实践 连载十九 Flask工厂模式集成使用Celery

发布时间: 2020-05-23 更新时间: 2021-11-30

Flask后端实践 Flask,Python,Celery 4.20 K 3 分钟 174

Flask后端实践 连载十九 Flask工厂模式集成使用Celery

tips:

  • 讲解Flask与Celery结合使用中遇到的各种问题解决方法
  • 本文基于python3编写
  • 代码仓库

项目场景

项目上有许多任务需要在后台处理,虽然可以使用异步线程来解决,但是无法及时获取到任务执行状态,有时任务执行失败,也无法及时获取到关键信息。
因此,采用Celery来重写相关的异步任务,方便管理和处理错误信息。

前置条件

  • 安装Celery pip install celery==4.3.0
  • 安装Flask pip install flask==1.0.3
  • 安装redis pip install redis==3.2.1

Flask与Celery结合简单使用

  • 新建celery_app.py
    from celery import Celery redis_config='redis://127.0.0.1:6379/0' celery_app = Celery(__name__, backend=redis_config, broker=redis_config) @celery_app.task def async_add(x, y): return x + y
  • 新建flask_app.py
    from flask import Flask from celery_app import async_add flask_app = Flask(__name__) @flask_app.route('/test', methods=["GET"]) def test_add(): """ 测试相加 :return: """ result = async_add.delay(1, 2) return str(result.get(timeout=1)) if __name__ == '__main__': flask_app.run()
  • 启动
    • windows shell中celery -A celery_app worker --pool=solo -l info
    • Linux shell中 celery -A celery_app worker -l info
    • 启动Flask APP python flask_app.py
  • 测试
    • 访问http://127.0.0.1:5000/test 返回 3

工厂模式下Flask与Celery结合

  • 文件目录结构

    _
     |_ app
     |  |_ __init__.py # 工程函数
     |  |_ api.py  # API接口
     |  |_ celery.py # celery及tasks
     |  
     |_ run.py # 启动Flask APP
    
    
  • 编写celery.py

    from celery import Celery from flask import current_app celery_app = Celery(__name__) @celery_app.task def add(x, y): """ 加法 :param x: :param y: :return: """ return str(x + y) @celery_app.task def flask_app_context(): """ celery使用Flask上下文 :return: """ with current_app.app_context(): return str(current_app.config)
  • 编写__init__.py

    from flask import Flask from .celery import celery_app def create_app(): app = Flask(__name__) celery_app.conf.update({"broker_url": 'redis://127.0.0.1:6379/0', "result_backend": 'redis://127.0.0.1:6379/0', }) from .api import bp app.register_blueprint(bp) return app
  • 编写api.py

    from flask import Blueprint from .celery import add, flask_app_context bp = Blueprint("test", __name__, url_prefix='/') @bp.route('/testAdd', methods=["GET"]) def test_add(): """ 测试相加 :return: """ result = add.delay(1, 2) return result.get(timeout=1) @bp.route('/testFlaskAppContext', methods=["GET"]) def test_flask_app_context(): """ 测试获取flask上下文 :return: """ result = flask_app_context.delay() return result.get(timeout=1).replace('<Config', '')
  • 编写run.py

    from app import create_app,celery_app app = create_app() # 关键点,往celery推入flask信息,使得celery能使用flask上下文 app.app_context().push() if __name__ == '__main__': app.run()
  • 启动

    • windows shell中输入celery -A run:celery_app worker --pool=solo -l info
    • Linux shell中输入celery -A run:celery_app worker -l info
    • 启动Flask APP python run.py
  • 测试

    • 访问http://127.0.0.1:5000/testAdd 返回 3
    • 访问http://127.0.0.1:5000/testFlaskAppContext 返回 Flask配置信息

总结

  • 本片文章主要解决了新版本的Celery使用Flask上下文的功能
end
如果你觉得还不错的话,请我吃个午饭吧!😍
支付宝
支付宝
微信
微信
目录

Copyright © 2019-2020 qzq版权所有

蜀ICP备19012274号-1 | 管理