当前位置: 首页>后端>正文

Flask框架——基于Celery的后台任务

上篇文章我们学习了Flask框架——MongoEngine使用MongoDB数据库,这篇文章我们学习Flask框架——基于Celery的后台任务。

Celery

在Web开发中,我们经常会遇到一些耗时的操作,例如:上传/下载数据、发送邮件/短信,执行各种任务等等。这时我们可以使用分布式异步消息任务队列去执行这些任务。

Celery是一款非常简单、灵活、可靠的分布式异步消息队列工具,可以用于处理大量消息、实时数据以及任务调度。

Celery通过消息机制进行通信,一般使用中间人(Broker)作为客户端和职程(Worker)调节。

其工作流程如下图所示:


Flask框架——基于Celery的后台任务,第1张

客户端发送消息任务给中间人(Broker),任务执行单元(Celery Worker)监控中间人中的任务队列,当中间人有消息任务时就分配任务给任务执行单元,任务执行单元在后台运行任务并返回请求。

注意:Celery可以有多个职程(Worker)和中间人(Broker),用来提高Celery的高可用性以及横向扩展能力。

Celery优点

  • 简单:上手比较简单,不需要配置文件就可以直接运行;
  • 高可用:如果出现丢失连接或连接失败,职程(Worker)和客户端会自动重试,并且中间人通过 主/主 主/从 的方式来进行提高可用性;
  • 快速:单个 Celery 进行每分钟可以处理数以百万的任务,而且延迟仅为亚毫秒(使用 RabbitMQ、 librabbitmq 在优化过后);
  • 灵活:Celery 的每个部分几乎都可以自定义扩展和单独使用,例如自定义连接池、序列化方式、压缩方式、日志记录方式、任务调度、生产者、消费者、中间人(Broker)等。

安装

Celery安装方式很简单,执行如下命令即可:

pip install celery

这里我们使用redis作为中间人,执行如下代码安装redis:

pip install redis

创建Celery程序

对比说明

(1)不使用Celery执行耗时任务,创建一个名为test.py文件,其示例代码如下:

import time

def add(a,b):                   
    time.sleep(5)               #休眠5秒
    return a+b      

if __name__ == '__main__':
    print('开始执行')
    result=add(2,3)             #调用add函数
    print('执行结束')
    print(result)

运行test.py文件,运行结果如下图:


Flask框架——基于Celery的后台任务,第2张

(2)使用Celery执行耗时任务,创建一个名为tasks.py文件,示例代码如下:

import time
from celery import Celery

celery = Celery(                          #实例化Celery对象
    'tasks',                            #当前模块名
    broker='redis://localhost:6379/1',      #使用redis为中间人
    backend='redis://localhost:6379/2'      #结果存储
)

@celery.task()              #使用异步任务装饰器task
def add(a,b):
    time.sleep(5)           #休眠5秒
    return a+b

if __name__ == '__main__':
    print('开始执行')
    result=add.delay(2,3)           #调用add方法并使用delay延时函数    
    print('执行结束')
    print(result)

实例化Celery对象,其中第一个参数为当前模块名,第二个参数为中间人(Broker)的URL链接,第三个参数为中间人结果放回的存储URL链接,再调用add()方法时,需要使用delay延时函数。

运行tasks.py文件,运行结果如下图所示:

Flask框架——基于Celery的后台任务,第3张

当我们运行tasks.py文件时,发现程序一下子就运行结束并返回任务id,

在终端执行如下代码运行Celery职程(Worker)服务:

celery -A tasks worker -l info

如下图所示:


Flask框架——基于Celery的后台任务,第4张

虽然职程已经收到任务并且在分配到子进程运行了,但是发现该任务没有运行结束,这时因为Celery不支持在windows下运行任务,需要借助eventlet来完成,执行如下安装eventlet:

pip install eventlet 

安装成功后,执行如下代码运行Celery职程(Worker)服务:

celery -A tasks worker -l info -P eventlet  -c 10

运行结果如下:


Flask框架——基于Celery的后台任务,第5张

Celery配置

大多数情况下,使用默认的配置即可满足我们的开发,不需要修改配置,当我们需要修改配置时,可以通过update进行配置,在上面的tasks.py添加如下代码:

celery.conf.update(
    task_serializer='json',
    accept_content=['json'],   
    result_serializer='json',
    timezone='Europe/Oslo',
    enable_utc=True,
)

其中:

  • accept_content:允许的内容类型/序列化程序的白名单,如果收到不在此列表中的消息,则该消息将被丢弃并出现错误,默认只为json;

  • task_serializer:标识要使用的默认序列化方法的字符串,默认值为json;

  • result_serializer:结果序列化格式,默认值为json;

  • timezone:配置Celery以使用自定义时区;

  • enable_utc:启用消息中的日期和时间,将转换为使用 UTC 时区,与timezone连用,当设置为 false 时,将使用系统本地时区。

除了上面的配置参数,Celery还提供了很多很多配置参数,大家可以在官方配置文档中查看

Celery的配置信息比较多,通常情况下,我们会在tasks.py同级目录下为创建Celery的配置文件, 这里命名为celeryconfig.py,示例代码如下:

broker_url = 'redis://localhost:6379/1'
result_backend = 'redis://localhost:6379/2'
task_serializer = 'json'
result_serializer = 'json'
accept_content = ['json']
timezone = 'Europe/Oslo'
enable_utc = True

执行如下代码加载配置:

import celeryconfig
app.config_from_object('celeryconfig')

在Flask项目中使用Celery

首先创建一个名为mycelery.py文件,该文件用来实例化Celery对象,示例代码如下:

from celery import Celery

def make_celery(app):
    celery = Celery(                        #实例化Celery
        'tasks',
        broker='redis://localhost:6379/1',      #使用redis为中间人
        backend='redis://localhost:6379/2'      #结果存储
    )
    class ContextTask(celery.Task):             #创建ContextTask类并继承Celery.Task子类
        def __call__(self, *args, **kwargs):    
            with app.app_context():                 #和Flask中的app建立关系
                return self.run(*args, **kwargs)    #返回任务
    celery.Task = ContextTask                   #异步任务实例化ContextTask
    return celery                               #返回celery对象

首先自定义一个名为make_celery()方法,该方法传入Flask程序中的app,在方法中实例化Celery,并创建一个名为ContextTask类用来和Flask中的app建立关系,最后返回celery。

创建名为tasks.py文件,该文件用来存放我们的耗时任务,示例代码如下:

import time
from app import celery
@celery.task            #使用异步任务装饰器task
def add(x, y):
    time.sleep(5)       #休眠5秒
    return x + y

这里我们通过休眠的方式来模拟耗时的下载任务。

Flask程序app.py文件示例代码如下:

from flask import Flask
import tasks
from mycelery import make_celery

app = Flask(__name__)
celery = make_celery(app)               #调用make_celery方法并传入app使celery和app进行关联

@app.route('/')
def hello():
    tasks.add.delay(1,2)                #调用tasks文件中的add()异步任务方法
    return '请求正在后台处理中,您可以去处理其他事情'

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

app.py文件很简单,就调用make_celery方法使celery和app进行关联,并在视图函数中使用tasks中的异步任务方法。

在终端执行如下代码运行Celery职程(Worker)服务:

celery -A tasks worker -l info -P eventlet  -c 10

启动Flask程序,访问http://127.0.0.1:5000/后在终端查Worker服务,如下图所示:

Flask框架——基于Celery的后台任务,第6张

这样就成功使用Celery把耗时任务交给后台来处理,避免了不必要的耗时等待(如下载数据任务)。

当我们不使用Celery时,用户在执行耗时任务时,用户可能要等耗时任务完成后,才能进行其他操作。

好了,Flask框架——基于Celery的后台任务就讲到这里了,感谢观看,下篇文章继续学习Flask框架的其他知识。
公众号:白巧克力LIN

该公众号发布Python、数据库、Linux、Flask、自动化测试、Git等相关文章!


https://www.xamrdz.com/backend/3n81931360.html

相关文章: