资讯

精准传达 • 有效沟通

从品牌网站建设到网络营销策划，从策略到执行的一站式服务

网站建设 >

查看其它板块

如何在flask中使用celery？

前言

在web开发中我们经常会遇到一些耗时的操作，比如发送邮件/短信，执行各种任务等等，这时我们会采取异步的方式去执行这些任务，而celery就是这样的一个异步的分布式任务处理框架，官方文档  
今天，我们的主题是celery如何与flask一起工作，我们都知道，flask是一个非常小巧的web框架，有许许多多的扩展，celery也不例外，我们先看下目前常用的几个flask-celery的扩展：

创新互联是一家集网站建设,吉水企业网站建设,吉水品牌网站建设,网站定制,吉水网站建设报价,网络营销,网络优化,吉水网站推广为一体的创新建站企业，帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿，时刻以成就客户成长自我，坚持不断学习、思考、沉淀、净化自己，让我们为更多的企业打造出实用型网站。

  1. Flask-Celery: celery作者本人开发的，其实不算扩展，功能就是安装celery及其相关组件，这里不谈。
  2. Flask-Celery-Helper：曾经的扩展，作者已不维护，不支持现在的4.0版本
  3. Flask-CeleryExt：支持4.0版本，目前比较好用的扩展

除这些扩展之外，其实flask的官方文档中已经给出了在flask中使用celery的方式，不过，那是一个单文件中运行flask的demo，在实际项目中使用，还是有许多需要注意的地方，接下来，我们就一起探究下如何在flask项目中使用celery。

项目结构

├── celery_task                   # celery任务相关
│   ├── __init__.py
│   ├── tasks.py
│   └── test.py
├── manage.py                     # celery worker实例
├── requirements.txt              # 依赖包
└── test_api                      # flask 项目
    ├── api                       # 蓝本相关
    │   ├── __init__.py
    │   └── v1
    │       ├── __init__.py
    │       └── views.py
    ├── extensions.py             # 扩展初始化
    ├── __init__.py               # flask app
    ├── models.py                 # 模型文件
    └── settings.py               # 配置文件

官方示例代码

本项目中没有使用扩展，只是基于官方文档中的示例做进一步的应用。

from celery import Celery

def make_celery(app):
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery

这是一个celery的工厂函数，使用flask app中的配置设置celery相关的属性，并且更改了celery对象的Task，使其能够使用flask的应用上下文，这一点非常重要。我们将这段代码放置到flask项目初始化文件中去也就是testapi/__init_\.py

构建celery对象

celerytask/__init_\.py

rom test_api import create_app, make_celery

app = create_app()
celery = make_celery(app)

class MyTask(celery.Task): # celery 基类

    def on_success(self, retval, task_id, args, kwargs):
        # 执行成功的操作
        print('MyTasks 基类回调，任务执行成功')
        return super(MyTask, self).on_success(retval, task_id, args, kwargs)

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # 执行失败的操作
        # 任务执行失败，可以调用接口进行失败报警等操作
        print('MyTasks 基类回调，任务执行失败')
        return super(MyTask, self).on_failure(exc, task_id, args, kwargs, einfo)

这里我对Task做了进一步的定制，用于添加一些任务信息。

编写任务

import datetime
import time
import os
import random
from flask import current_app
from test_api.models import User
from test_api.extensions import db

from celery_task import celery, MyTask

@celery.task(bind=True, base=MyTask)
def apptask(self):
    print(current_app.config)
    print("==============%s " % current_app.config["SQLALCHEMY_DATABASE_URI"])
    print("++++++++++++++%s " % os.getenv("DATABASE_URL"))
    time.sleep(5)
    user = User(username="user%s" % random.randint(1,100))
    db.session.add(user)
    db.session.commit()
    return 'success'

这个任务很简单，使用User模型类异步向数据库中添加数据，为了体现耗时操作，使用sleep函数模拟。

视图函数中使用

test_api/api/v1/views.py

from flask import jsonify
from celery_task.tasks import apptask
from test_api.api.v1 import api_v1
from test_api.extensions import db
from flask import current_app

@api_v1.route("/", methods=["GET"])
def index():
    r = apptask.apply_async()
    return jsonify({"status": "success"})

视图函数非常的简单，只做了提交任务的操作。

启动并测试

启动celery

为了避免循环导入问题，我们在项目根目录下新建manage.py

from test_api import create_app, make_celery

app = create_app()
celery = make_celery(app)

if __name__ == '__main__':
    app.run()

这个文件只用来启动celery，启动命令如下:

# celery worker -A manage:celery -l debug

看到如下输出，表明启动成功：

-------------- celery@test-3 v4.4.0 (cliffs)
--- ***** ----- 
-- ******* ---- Linux-3.10.0-693.2.2.el7.x86_64-x86_64-with-centos-7.4.1708-Core 2020-03-03 21:14:13
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         test_api:0x7f87c31a4e48
- ** ---------- .> transport:   redis://127.0.0.1:6379/3
- ** ---------- .> results:     redis://127.0.0.1:6379/4
- *** --- * --- .> concurrency: 2 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

[tasks]
  . celery.accumulate
  . celery.backend_cleanup
  . celery.chain
  . celery.chord
  . celery.chord_unlock
  . celery.chunks
  . celery.group
  . celery.map
  . celery.starmap
  . celery_task.tasks.apptask

[2020-03-03 21:14:13,632: DEBUG/MainProcess] | Worker: Starting Hub
[2020-03-03 21:14:13,632: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:13,632: DEBUG/MainProcess] | Worker: Starting Pool
[2020-03-03 21:14:13,690: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:13,691: DEBUG/MainProcess] | Worker: Starting Consumer
[2020-03-03 21:14:13,691: DEBUG/MainProcess] | Consumer: Starting Connection
[2020-03-03 21:14:13,708: INFO/MainProcess] Connected to redis://127.0.0.1:6379/3
[2020-03-03 21:14:13,708: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:13,708: DEBUG/MainProcess] | Consumer: Starting Events
[2020-03-03 21:14:13,718: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:13,718: DEBUG/MainProcess] | Consumer: Starting Mingle
[2020-03-03 21:14:13,718: INFO/MainProcess] mingle: searching for neighbors
[2020-03-03 21:14:14,743: INFO/MainProcess] mingle: all alone
[2020-03-03 21:14:14,743: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:14,744: DEBUG/MainProcess] | Consumer: Starting Gossip
[2020-03-03 21:14:14,748: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:14,748: DEBUG/MainProcess] | Consumer: Starting Heart
[2020-03-03 21:14:14,750: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:14,750: DEBUG/MainProcess] | Consumer: Starting Tasks
[2020-03-03 21:14:14,756: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:14,756: DEBUG/MainProcess] | Consumer: Starting Control
[2020-03-03 21:14:14,759: DEBUG/MainProcess] ^-- substep ok
[2020-03-03 21:14:14,759: DEBUG/MainProcess] | Consumer: Starting event loop
[2020-03-03 21:14:14,759: DEBUG/MainProcess] | Worker: Hub.register Pool...
[2020-03-03 21:14:14,760: INFO/MainProcess] celery@test-3 ready.
[2020-03-03 21:14:14,760: DEBUG/MainProcess] basic.qos: prefetch_count->8

启动flask:

# flask run

* Serving Flask app "test_api" (lazy loading)
 * Environment: development
 * Debug mode: on
 * Running on http://0.0.0.0:5000/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 237-492-852

调试接口:

# curl http://127.0.0.1:5000/api/v1/
{
  "status": "success"
}

查看celery日志：

[2020-03-03 21:17:31,330: WARNING/ForkPoolWorker-2] 
[2020-03-03 21:17:31,330: DEBUG/MainProcess] Task accepted: celery_task.tasks.apptask[5f27a148-161f-4485-931f-17d94637168e] pid:2341
[2020-03-03 21:17:36,391: WARNING/ForkPoolWorker-2] MyTasks 基类回调，任务执行成功
[2020-03-03 21:17:36,392: INFO/ForkPoolWorker-2] Task celery_task.tasks.apptask[5f27a148-161f-4485-931f-17d94637168e] succeeded in 5.0624741315841675s: 'success'

任务执行成功，查看数据库数据：

MySQL> select * from user order by id;
+----+----------+
| id | username |
+----+----------+
|  1 | user26   |
|  2 | user69   |
|  3 | user71   |
|  4 | user35   |
|  5 | user13   |
|  6 | user54   |
|  7 | user88   |
|  8 | user63   |
|  9 | user87   |
| 10 | user90   |
| 11 | user3    |
| 12 | user18   |
| 13 | user65   |
+----+----------+

数据已被插入，实验成功!

总结

有几个坑希望大家注意下

1. app初始化文件中蓝图导入位置问题引起循环导入，导致import Error

出错文件： testapi/__init_\.py

import os 
import click

from flask import Flask, jsonify
from test_api.api.v1 import api_v1   # 蓝图在上方导入，循环报错产生
from test_api.settings import config
from test_api.models import User

from celery import Celery

def make_celery(app):
...
def create_app(config_name=None):
    if config_name is None:
        config_name = os.getenv('FLASK_ENV', 'development')

    app = Flask('test_api')
    app.config.from_object(config[config_name])

    register_extensions(app)
    register_blueprints(app)
    register_commands(app)
    register_errors(app)
    return app

# 注册蓝图函数
def register_blueprints(app):
    app.register_blueprint(api_v1, url_prefix='/api/v1')

启动celery和请求接口时均会报错，错误堆栈如下：

  from test_api import create_app, make_celery
  File "/tmp/test/test_api/__init__.py", line 5, in 
    from test_api.api.v1 import api_v1
  File "/tmp/test/test_api/api/v1/__init__.py", line 9, in 
    from test_api.api.v1 import views
  File "/tmp/test/test_api/api/v1/views.py", line 2, in 
    from celery_task.tasks import apptask
  File "/tmp/test/celery_task/__init__.py", line 1, in 
    from test_api import create_app, make_celery
ImportError: cannot import name 'create_app'
解决方法：

将蓝图的导入下放置蓝图注册函数中testapi/__init_\.py：

...
def register_blueprints(app):
    from test_api.api.v1 import api_v1
    app.register_blueprint(api_v1, url_prefix='/api/v1')
...
2. celery无法读取到flask-sqlalemy的连接配置信息

提交任务，celery报错如下：

  ...
  options = self.get_options(sa_url, echo)
  File "/tmp/py3/lib/python3.6/site-packages/flask_sqlalchemy/__init__.py", line 575, in get_options
    self._sa.apply_driver_hacks(self._app, sa_url, options)
  File "/tmp/py3/lib/python3.6/site-packages/flask_sqlalchemy/__init__.py", line 877, in apply_driver_hacks
    if sa_url.drivername.startswith('mysql'):
AttributeError: 'NoneType' object has no attribute 'drivername'

通过调试我发现，flask的app的配置是可以拿到的，因为我们在工厂函数中推送了应用上下文，我的数据库配置信息是以键值的形式写在了.env文件中，这也是目前flask推荐的方式。那为什么celery取不到数据库连接配置呢？其实，启动celery的app和我们web服务所用app是两个独立的app，celery无法通过.env中的环境变量取到相应的值，这里有三种解决办法：

  • 不使用环境变量的方式，直接将相关信息写在配置文件中例如: SQLALCHEMY_DATABASE_URI = "mysql+pymysql://xxx:xxx@127.0.0.1:3306/test?charset=utf8"

  • 将配置写到系统环境变量中（/etc/profile）
  • 使用dotenv加载.env中的环境变量

相比之下，方案三是采纳比较多的，于是我们在test_api/settings.py文件中加入如下代码:

from dotenv import find_dotenv, load_dotenv
load_dotenv(find_dotenv())

find_dotenv函数会在当前以及父目录中搜寻.env文件，load_dotenv函数则负责加载环境变量。如此，大功告成。我们可以继续愉快撸代码啦。  
附：项目源码


网站题目：如何在flask中使用celery？
URL标题：http://www.cdkjz.cn/article/pjjegs.html
返回首页 了解更多建站资讯

相关资讯

多年建站经验

多一份参考，总有益处

联系快上网，免费获得专属《策划方案》及报价

咨询相关问题或预约面谈，可以通过以下方式与我们联系

大客户专线   成都：13518219792   座机：028-86922220

在线咨询 提交需求
友情链接 交换友情链接
成都气球装饰香港虚拟主机空间温江网站建设电商网站建设软文营销网站开发西部信息中心成都小程序开发公司成都展柜设计四川成都网站制作

成都网站建设公司地址：成都市青羊区太升南路288号锦天国际A座10层 建设咨询028-86922220

成都快上网科技有限公司-四川网站建设设计公司 | 蜀ICP备19037934号 Copyright 2020,ALL Rights Reserved cdkjz.cn | 成都网站建设 | © Copyright 2020版权所有.

专家团队为您提供成都网站建设,成都网站设计,成都品牌网站设计,成都营销型网站制作等服务,成都建网站就找快上网！ | 成都网站建设哪家好？ | 网站建设地图