flask 工厂模式与celery结合

Posted 2020-12-20 se7enjean

前言

简单介绍一下celery：Celery 是一个异步任务队列。你可以使用它在你的应用上下文之外执行任务。总的想法就是你的应用程序可能需要执行任何消耗资源的任务都可以交给任务队列，让你的应用程序自由和快速地响应客户端请求。
官方文档：https://docs.celeryproject.org/en/stable/
中文文档：https://www.celerycn.io/ru-men/celery-jian-jie
Flask 中使用 Celery：http://www.pythondoc.com/flask-celery/first.html

安装

因为以下操作在flask中执行，根据网上说的，celery最新版不支持windows系统，所以我这边安装的是v3.0版本，最新版还未尝试，感兴趣的同学可以自行操作一番。
中间人采用的是redis，这里选择低版本的redis安装使用，不然会报一个不兼容的错误。

# 推荐使用国内镜像源安装
pip install flask
pip install celery==3.1.25 
pip install redis==2.10.6

首次体验

当安装上面的包后，我们先在最基础的flask框架下试一下，能否启动成功。

代码实现

这是flask初始化的app.py

from flask import Flask

app = Flask(__name__)


@app.route(‘/‘)
def hello_world():
    return ‘Hello World!‘


if __name__ == ‘__main__‘:
    app.run(debug=True)

配置后的app.py

import time

from flask import Flask
from celery import Celery

app = Flask(__name__)

# 这里是配置中间人redis基础写法，可以照抄。如果是服务器，需要更换（127.0.0.1）
app.config[‘CELERY_BROKER_URL‘] = ‘redis://127.0.0.1:6379/0‘
app.config[‘CELERY_RESULT_BACKEND‘] = ‘redis://127.0.0.1:6379/0‘

# 初始化celery
celery_ = Celery(app.name, backend=app.config[‘CELERY_RESULT_BACKEND‘], broker=app.config[‘CELERY_BROKER_URL‘])
celery_.conf.update(app.config)


@app.route(‘/‘)
def hello_world():
    return ‘Hello World!‘


# 这里定义后台任务，异步执行装饰器@celery_.task
@celery_.task
def my_background_task(arg1):
    # some long running task here
    time.sleep(5)
    result = arg1
    print(result.id)
    return result


# 以下是两个视图，测试celery能否正常使用
@app.route("/index")
def index():
    """一个测试的实例"""
    # print(my_background_task(3)) # add函数也能做普通的函数使用
    result = my_background_task.apply_async(args=[5, ])  # 发送异步任务，指定队列

    return result.id


@app.route("/index/<result_id>")
def get_result(result_id):
    # 根据任务ID获取任务结果

    print(result_id)
    # 这里请注意，网上很多教程只有“AsyncResult”，但这是不对的
    # 这里要调用的是实例化后的celery的对象方法才行，如果不是这样，启动celery服务后会报一个配置后端错误
    result = celery_.AsyncResult(id=result_id)
    print(result)
    return str(result.get())


if __name__ == ‘__main__‘:
    app.run(debug=True)

启动服务

我们尝试在pycharm的窗口下执行命令

注意，一下窗口是另起的一个窗口用来启动celery服务。

可以通过访问http://127.0.0.1:5000/index查看

结合flask

先介绍一下我的工程目录结构，大家根据自己实际情况去设计，毕竟我还是比较菜的，学我可能会误人子弟，哈哈。

init.py

from celery import Celery

celery = Celery(__name__, broker=‘redis://127.0.0.1:6379/0‘, backend=‘redis://127.0.0.1:6379/0‘)

tasks.py

import time

from . import celery


@celery.task()
def add(x, y):
    """
    加法
    :param x:
    :param y:
    :return:
    """
    time.sleep(10)
    return str(x + y)

applocation.py

from flask import Flask
from celery_task import celery

def create_app(config_name):
    """
    创建flask应用对象
    :param config_name: str 配置模式的模式名字 （“develop”,“product”）
    :return:
    """
    app = Flask(__name__)

    """
    只需要在初始化app中加入这行代码，将下面的配置信息写入app的配置文件
    app.config[‘CELERY_BROKER_URL‘] = ‘redis://127.0.0.1:6379/0‘
    app.config[‘CELERY_RESULT_BACKEND‘] = ‘redis://127.0.0.1:6379/0‘
    """
    
    # 添加实例化需要在app实例之后
    celery.conf.update(app.config)

    # .........
    
    return app

以上的方式是为了解决循环导入的问题，如果大家有更好的方案，可以优化一下。
接下来的启动方式与上面一样，开两个窗口，一个启动flask项目，一个启动celery服务。
！！！注意！！！