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

Celery

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

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

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

其工作流程如下图所示：

客户端发送消息任务给中间人（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文件，运行结果如下图：

（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文件，运行结果如下图所示：

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

在终端执行如下代码运行Celery职程（Worker）服务：

celery -A tasks worker -l info

如下图所示：

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

pip install eventlet 

安装成功后，执行如下代码运行Celery职程（Worker）服务：

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

运行结果如下：

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服务，如下图所示：

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

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

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