用 Flask 来写个轻博客 (26) — 使用 Flask-Celery-Helper 实现异步任务

简介: 目录目录前文列表扩展阅读Celery将 Celery 加入到应用中实现向新用户发送欢迎邮件前文列表用 Flask 来写个轻博客 (1) — 创建项目 用 Flask 来写个轻博客 (2) — Hello World! 用 Flask 来...

目录

前文列表

用 Flask 来写个轻博客 (1) — 创建项目
用 Flask 来写个轻博客 (2) — Hello World!
用 Flask 来写个轻博客 (3) — (M)VC_连接 MySQL 和 SQLAlchemy
用 Flask 来写个轻博客 (4) — (M)VC_创建数据模型和表
用 Flask 来写个轻博客 (5) — (M)VC_SQLAlchemy 的 CRUD 详解
用 Flask 来写个轻博客 (6) — (M)VC_models 的关系(one to many)
用 Flask 来写个轻博客 (7) — (M)VC_models 的关系(many to many)
用 Flask 来写个轻博客 (8) — (M)VC_Alembic 管理数据库结构的升级和降级
用 Flask 来写个轻博客 (9) — M(V)C_Jinja 语法基础快速概览
用 Flask 来写个轻博客 (10) — M(V)C_Jinja 常用过滤器与 Flask 特殊变量及方法
用 Flask 来写个轻博客 (11) — M(V)C_创建视图函数
用 Flask 来写个轻博客 (12) — M(V)C_编写和继承 Jinja 模板
用 Flask 来写个轻博客 (13) — M(V)C_WTForms 服务端表单检验
用 Flask 来写个轻博客 (14) — M(V)C_实现项目首页的模板
用 Flask 来写个轻博客 (15) — M(V)C_实现博文页面评论表单
用 Flask 来写个轻博客 (16) — MV(C)_Flask Blueprint 蓝图
用 Flask 来写个轻博客 (17) — MV(C)_应用蓝图来重构项目
用 Flask 来写个轻博客 (18) — 使用工厂模式来生成应用对象
用 Flask 来写个轻博客 (19) — 以 Bcrypt 密文存储账户信息与实现用户登陆表单
用 Flask 来写个轻博客 (20) — 实现注册表单与应用 reCAPTCHA 来实现验证码
用 Flask 来写个轻博客 (21) — 结合 reCAPTCHA 验证码实现用户注册与登录
用 Flask 来写个轻博客 (22) — 实现博客文章的添加和编辑页面
用 Flask 来写个轻博客 (23) — 应用 OAuth 来实现 Facebook 第三方登录
用 Flask 来写个轻博客 (24) — 使用 Flask-Login 来保护应用安全
用 Flask 来写个轻博客 (25) — 使用 Flask-Principal 实现角色权限功能

扩展阅读

Celery-分布式任务队列
基于后台作业的 Celery
Flask-Celery-Helper 1.1.0

Celery

Celery 是使用 Python 多任务库来编写的任务队列工具, 可以 并行 的执行任务. 我们会将执行时间较长但又不那么追求实时的功能以异步任务的形式完成, EG. 上传文件, 发送邮件…, Python 和 Celery 之间需要一个中间人(消息队列)来进行任务队列的管理, Celery 官方推荐使用 RabbirMQ 或 Redis 来充当这个角色. 当然也可以同时使用两者, 其中 MQ 作为中间人, Redis 传递 Celery 执行的结果给应用端. 这样做的优势在于, 返回给应用的结果是持久化保存在数据库中的.

消息队列: 是一种专门设计的系统, 用于在生产者(往队列发送消息的程序)和消费者(从队列中取出消息的队列)之间传递消息.

这里写图片描述

  • 安装 Celery
pip install Celery
  • 安装 Flask-Celery-Helper
    Flask-Celery-Helper 是一个 Flask 扩展, 用于辅助使用 app 来初始化 Celery 对象, 使其得以注册到 app 对象中.
pip install Flask-Celery-Helper
pip freeze > requirments.txt
/etc/init.d/rabbitmq-server start

将 Celery 加入到应用中

  • 配置 Celery 连接 RabbitMQ 的 URL
    vim jmilkfansblog/config.py
class DevConfig(Config):
    """Development config class."""
...
    # Celery <--> RabbitMQ connection
    CELERY_RESULT_BACKEND = "amqp://guest:guest@localhost:5672//"
    CELERY_BROKER_URL = "amqp://guest:guest@localhost:5672//"

NOTE: RabbitMQ 使用默认的 guest 用户, 端口为 5672

  • 创建 celery 对象
    vim jmilkfansblog/extensions.py
from flask.ext.celery import Celery

...

# Create the Flask-Celery-Helper's instance
flask_celery = Celery()
  • 将 celery 对象注册到 app 对象
    vim jmilkfansblog/__init__.py
from jmilkfansblog.extensions import flask_celery

def create_app(object_name):
    """Create the app instance via `Factory Method`"""
...
    # Init the Flask-Celery-Helper via app object
    # Register the celery object into app object
    flask_celery.init_app(app)

NOTE 1: Celery 的进程必须在 Flask app 的上下文中运行, 这样 Celery 才能够跟其他的 Flask 扩展协同工作。所以必须将 flask_celery 对象注册到 app 对象中, 并且每创建一个 Celery 进程都需要创建一个新的 Flask app 对象, 这里我们使用工厂模式来创建 celery 对象。这一点是非常重要的,实际上 Flask application 和 Celery application 是两个不同的进程,在 Celery 没有加入 Flask 上下文的情况下,Celery 的程序逻辑就不能轻易的访问 Flask 相关资源,比如不能加载 Flask 的环境配置信息,无法通过 Flask 来访问数据库,不能使用 Flask 的扩展功能等。如果想做到这些,Celery 都需要自己再实现一套相同的逻辑,这样做显然是没有必要的。所以 Flask application 原生支持将自己的 Context 嵌入到别的 application 中,当然有些情况也需要相应扩展的辅助,例如 Flask-Celery-Helper 在这里就充当着这个角色。

NOTE 2: flask_celery 对象是 Flask-Celery-Helper 扩展的对象, 用于辅助处理 Celery 的初始化, 所以实际上我们是可以不使用这个扩展, 而直接使用 Celery 的. celery 对象才是真正的 Celery 的对象.

  • 使用工厂模式来创建 celery 对象
    ./celery_runner.py
mport os

from celery import Celery

from jmilkfansblog import create_app


def make_celery(app):
    """Create the celery process."""

    # Init the celery object via app's configuration.
    celery = Celery(
        app.import_name,
        backend=app.config['CELERY_RESULT_BACKEND'],
        broker=app.config['CELERY_BROKER_URL'])

    # Flask-Celery-Helpwe to auto-setup the config.
    celery.conf.update(app.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):

        abstract = True

        def __call__(self, *args, **kwargs):
            """Will be execute when create the instance object of ContextTesk."""

            # Will context(Flask's Extends) of app object(Producer Sit) 
            # be included in celery object(Consumer Site).
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    # Include the app_context into celery.Task.
    # Let other Flask extensions can be normal calls.
    celery.Task = ContextTask
    return celery

env = os.environ.get('BLOG_ENV', 'dev')
flask_app = create_app('jmilkfansblog.config.%sConfig' % env.capitalize())
# 1. Each celery process needs to create an instance of the Flask application.
# 2. Register the celery object into the app object.
celery = make_celery(flask_app)

NOTE 1: 我们以后会以 CLI 的形式来管理和控制 Celery 的 worker, 所以我们将 celery 对象的实现模块放置在 ./celery_runner.py 中, 而不是 jmilkfansblog/celery_runner.py. Flask app 内部的 tasks 任何的定义和实现就交由 Flask-Celery-Helper 来支持就好了, 这也是 Flask-Celery-Helper 存在的意义.

NOTE 2: make_celery()最重要的作用就是让每个 Celery 的进程中(celery对象)都包含有 app 对象的上下文, 至于为什么这么做呢? 上述已经给出了答案.

NOTE 3: 这里通过 create_app() 创建的对象不能够命名为 app, 而是命名为 flask_app, 这是因为 Celery 会默认将命名为 app 或 celery 的对象都作为一个 Celery 对象来处理.

NOTE 4: celery.conf.update(app.config) 会将 app 对象的 config 更新到 celery 对象中, 当然也包括了刚刚定义的 RabbitMQ 连接 URL 配置.

  • 启动 Celery 服务
(env) jmilkfan@JmilkFan-Devstack:/opt/JmilkFan-s-Blog$ celery worker -A celery_runner --loglevel=info
...

 -------------- celery@JmilkFan-Devstack v4.0.1 (latentcall)
---- **** ----- 
--- * ***  * -- Linux-4.4.0-53-generic-x86_64-with-Ubuntu-16.04-xenial 2016-12-15 19:12:33
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         jmilkfansblog:0x7f5b24345990
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     amqp://
- *** --- * --- .> concurrency: 4 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . jmilkfansblog.tasks.digest
  . jmilkfansblog.tasks.log
  . jmilkfansblog.tasks.multiply
  . jmilkfansblog.tasks.remind

NOTE: 在启动 Celery 服务的时候, 能够用 Output 看见其自身的配置信息和现在所处理的 tasks.

实现向新用户发送欢迎邮件

下面使用 Celery 实现在用户创建账户之后, 在指定的时间内异步的向新用户发送欢迎邮件.

  • 添加 Reminder Model 用户存放用户的 email 地址和欢迎内容.
    vim jmilkfansblog/models.py
class Reminder(db.Model):
    """Represents Proected reminders."""

    __tablename__ = 'reminders'
    id = db.Column(db.String(45), primary_key=True)
    date = db.Column(db.DateTime())
    email = db.Column(db.String(255))
    text = db.Column(db.Text())

    def __init__(self, id, text):
        self.id = id
        self.email = text

    def __repr__(self):
        return '<Model Reminder `{}`>'.format(self.text[:20])
  • 创建一个 task
    vim jmilkfansblog/tasks.py
import smtplib
import datetime
from email.mime.text import MIMEText

from flask_mail import Message

from jmilkfansblog.extensions import flask_celery, mail

@flask_celery.task(
    bind=True,
    igonre_result=True,
    default_retry_delay=300,
    max_retries=5)
def remind(self, primary_key):
    """Send the remind email to user when registered.
       Using Flask-Mail.
    """

    reminder = Reminder.query.get(primary_key)

    msg = MIMEText(reminder.text)
    msg['Subject'] = 'Welcome!'
    msg['FROM'] = <your_email>
    msg['To'] = reminder.email

    try:
        smtp_server = smtplib.SMTP('localhost')
        smtp_server.starttls()
        smtp_server.login(<user>, <password>)
        smtp_server.sendmail(<your_email>,
                             [reminder.email],
                             msg.as_string())
        smtp_server.close()
        return

    except Exception as err:
        self.retry(exc=err)

def on_reminder_save(mapper, connect, self):
    """Callbask for task remind."""
    remind.apply_async(args=(self.id), eta=self.date)

NOTE 1: Celery Task 本质上就是一个被 celery.task()装饰过的函数,

NOTE 2: 使用主键 primary_key 来获取 reminder 对象是为了避免数据竞态的发生, 因为从生成 reminder 对象到 task 被执行的过程并不能保证数据是最新的, 这也是处理异步调用时, 需要时刻注意的地方.

NOTE 3: on_reminder_save() 是一个回调函数, 当我们在一个特定的情景下调用这个函数的时候就触发了一个 Celery task.

  • 应用 SQLAlchemy 的 event 特性来触发 Celery task
    vim jmilkfansblog/__init__.py
from sqlalchemy import event


def create_app(object_name):
    """Create the app instance via `Factory Method`"""
...
    # Will be callback on_reminder_save when insert recond into table `reminder`.
    event.listen(Reminder, 'after_insert', on_reminder_save)
  • NOTE 1: SQLAlchemy 允许在 Model 上注册回调函数, 当 Model 对象发生特定的情景时, 就会执行这个回调函数, 这就是所谓的 event, 这里我们使用 after_insert 来指定当创建一个新的 Reminder 对象(插入一条记录)时就触发这个回调函数. 而是回调函数中的形参, 会由 event 来负责传入.
相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
6月前
|
网络协议 API Go
Flask 作者 Armin Ronacher:我不觉得有异步压力
Flask 作者 Armin Ronacher:我不觉得有异步压力
81 0
|
Java 测试技术 Python
Python:Flask使用ThreadPoolExecutor执行异步任务
Python:Flask使用ThreadPoolExecutor执行异步任务
699 0
|
3月前
|
NoSQL Redis UED
揭秘!Flask如何携手Celery,让异步任务处理不再是难题,打造极速响应的Web应用新纪元!
【8月更文挑战第31天】在Web开发中,Flask与Celery的结合为异步任务处理提供了强大支持。Flask作为轻量级Web框架,以其简洁灵活著称;而Celery则是一个分布式任务队列系统,擅长处理耗时任务。二者结合,Flask专注于处理Web请求,Celery则在后台异步执行复杂任务,如发送邮件或调用外部API。这种方式不仅提升了应用性能和响应速度,还优化了用户体验。下面的示例展示了如何在Flask应用中集成Celery以实现异步任务处理。
123 0
|
2月前
|
NoSQL 大数据 Redis
使用 Flask 和 Celery 构建异步任务处理
使用 Flask 和 Celery 构建异步任务处理
|
3月前
|
数据库 关系型数据库 MySQL
惊!Hibernate与MySQL的绝密优化技巧大揭秘,让你的数据库飞起来!
【8月更文挑战第31天】在企业应用开发中,结合使用持久层框架Hibernate与数据库管理系统MySQL可显著提升数据库交互效率。本文探讨了多项优化策略,包括配置二级缓存、采用单向关联减少JOIN操作、优化HQL查询语句以及合理使用MySQL索引。通过具体示例,文章详细讲解了如何实施这些优化措施,以期为企业应用提供更高效稳定的数据支持。
55 0
|
6月前
|
监控 NoSQL 测试技术
python使用Flask,Redis和Celery的异步任务
python使用Flask,Redis和Celery的异步任务
|
消息中间件 前端开发 NoSQL
Win11环境下使用Flask配合Celery异步推送实时/定时消息(Socket.io)
一般情况下,Celery被用来处理耗时任务,比如千篇一律的发邮件或者文件上传之类,本次使用Celery实时或者定时发送基于Websocket的消息队列,因为如果前端已经摒弃老旧的轮询策略,使用Websocket,后端则需要相应的配合Celery进行对持久化的Websocket链接主动推送消息,这种场景在生产环境中还是很常见的,但是网上却鲜有文章阐述,而Celery官方对此的说明是
Win11环境下使用Flask配合Celery异步推送实时/定时消息(Socket.io)
|
存储 前端开发 JavaScript
基于Flask开发网站 -- 前端Ajax异步上传文件到后台
从网页界面(前端)上传文件到服务器(后端)
462 0
基于Flask开发网站 -- 前端Ajax异步上传文件到后台
|
数据采集 SQL 数据可视化
flask + pyecharts 搭建新冠肺炎疫情数据可视化交互分析平台:包含疫情数据获取、态势感知、预测分析、舆情监测等任务
该项目是浙江大学地理空间数据库课程作业8:空间分析中,使用 flask + pyecharts 搭建的简单新冠肺炎疫情数据可视化交互分析平台,包含疫情数据获取、态势感知、预测分析、舆情监测等任务
2790 0
flask + pyecharts 搭建新冠肺炎疫情数据可视化交互分析平台:包含疫情数据获取、态势感知、预测分析、舆情监测等任务