August Rush

一个还在努力成长的小火汁!

游龙当归海,海不迎我自来也。

We create our own demons.

You can reach me at augustrush0923@gmail.com
Celery之初体验
发布:2021年03月24日 | 作者:augustrush | 阅读量: 1269

Celery介绍

Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题,使用非常简单。

Celery适用异步处理问题,当发送邮件、或者文件上传,图像处理等等一些比较耗时的操作,我们可将其异步执行,这样用户不需要等待很久,提高用户体验。

Celery的特点是:

  • 简单,易于使用和维护,有丰富的文档。
  • 高效,单个celery进程每分钟可以处理数百万个任务。
  • 灵活,celery中几乎每个部分都可以自定义扩展。


任务队列

任务队列是一种跨线程、跨机器工作的一种机制。

任务队列中包含称作任务的工作单元。有专门的工作进程持续不断的监视任务队列,并从中获得新的任务并处理。

celery通过消息进行通信,通常使用一个叫Broker(中间人)来协client(任务的发出者)worker(任务的处理者)clients发出消息到队列中,broker将队列中的信息派发给worker来处理。

一个celery系统可以包含很多的workerbroker,可增强横向扩展性和高可用性能。


安装

通过Python的pip安装或通过源码进行安装Celery。

  • 使用pip进行安装
pip install -U Celery
  • 源码安装
# https://pypi.org/project/celery/
$ tar xvfz celery-0.0.0.tar.gz
$ cd celery-0.0.0
$ python setup.py build
# python setup.py install


Broker

Celery需要一种解决消息的发送和接受的方式,我们把这种用来存储消息的中间装置叫做message broker,也可以叫做消息中间人。

这是不同的中间件比对情况,更多的信息可以在每个中间件的文档中找到(见众人指南)。

名称 状态 监控 远程控制
RabbitMQ 稳定
Redis 稳定
Amazon SQS 稳定
Zookeeper 实验阶段

目前实验阶段的中间人(Broker)只是功能性的,但是没有专门的维护人员。

缺少监控就意味着这个监控已经失效,因此相关的 Flower、Celery events、celerymon 和其他基于此功能的监控工具全部失效。

远程管理控制是指可以通过 celery inspect 和 celery control(以及使用远程控制API的工具)在程序运行时检查和管理职程(Worker)的能力。


应用

使用celery第一件要做的最为重要的事情是需要先创建一个Celery实例,我们一般叫做celery应用,或者更简单直接叫做一个app。app应用是我们使用celery所有功能的入口,比如创建人物,管理任务等,在使用celery的时候,app必须能够被其他的模块导入。


创建应用

首先创建一个名为tasks.py的文件,其内容为:

import time
from celery import Celery

# 使用redis作为broker,backend暂时为none
app = Celery('demo', broker='redis://@localhost:6357/1')

# 创建任务函数
@app.task
def my_task():
    print("任务正在执行...")
    time.sleep(3)
    print("任务结束执行...")

Celery第一个参数是给其设定一个名字。

Celery第二个参数设定一个中间人broker,在这里使用Redis作为中间人。对于RabbitMQ可以写为amqp://localhost, 使用Redis可以写为redis://localhost

my_task函数是我们编写的一个任务函数,通过加上装饰器app.task,将其注册到broker的队列中。


运行celery职程(worker)服务

现在我们在创建一个worker,等待处理队列中的任务。

# 在终端,cd到tasks.py同级目录中,执行命令
celery -A tasks worker --loglevel=info

显示效果如下:


调用任务

任务加入到broker队列中,以便刚才创建的celery worker服务器能够从队列中取出任务并执行。

需要调用我们创建的实例任务,可以通过delay()进行调用。

from tasks import task1

task1.delay()

该任务已经有职程(Worker)开始处理,可以通过控制台输出的日志进行查看执行情况。

调用任务会返回一个 AsyncResult 的实例,用于检测任务的状态,等待任务完成获取返回值(如果任务执行失败,会抛出异常)。默认这个功能是不开启的,如果开启则需要配置 Celery 的结果后端


存储结果

如果我们想跟踪任务的状态,Celery需要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、Redis、RPC(RabbitMQ/AMQP)以及自定义的后端结果存储中间件。

例子使用Redis作为存储结果的方案,任务结果存储配置我们通过Celery的backend参数来设定。

app = Celery('demo',
             broker='redis://@localhost:6379/1',
             backend='redis://@localhost:6379/2')

任务函数:

@app.task
def my_task(a, b):
    print("任务函数正在执行...")
    return a + b

我们给Celery增加了backend参数,指定redis作为结果存储,并将任务函数修改为两个参数,并且有返回值。

配置结果后端后,调用执行任务。会得到调用任务后返回的一个AsyncResult实例:

result = add.delay(10, 20)

ready()可以检测是否已经处理完毕:

result.ready()
# False

整个任务执行过程为异步的,如果一直等待任务完成,会将异步调用转换为同步调用:

result.get(timeout=3)
# 30

如果任务出现异常,get()会再次引发异常,可以通过propagate参数进行覆盖:

result.get(propagate=False)

如果任务出现异常,可以通过以下命令进行回溯:

result.traceback

完整的结果对象:

result.result

如果后端使用资源进行存储结果,必须要针对调用任务后返回每一个 AsyncResult 实例调用 get() 或 forget() ,进行资源释放。


配置

Celery使用简单,配置也非常简单。Celery有很多配置选项能够使得Celery能够符合我们的需要,但是默认的几项配置已经足够应付大多数应用场景了。

配置信息可以直接在app中设置, 或者通过专有的配置模块来配置。


直接通过app来配置

from celery import Celery

app = Celery('demo')

# 增加配置
app.conf.update(
    result_backend='redis://@localhost:6379/2',
    broker_url='redis://@localhost:6379/1'
)


专有配置文件

对于比较大的项目,建议配置信息作为一个单独的模块。我们可以通过调用app的函数来告诉Celery使用配置模块。

配置模块的名字我们取名为celeryconfig,这个名字不是固定的,我们可以任意取名,建议这么做我们必须保证配置模块能够被导入。

task.py文件同级目录下创建配置celeryconfig.py

result_backend = 'redis://:@127.0.0.1:6379/2'
broker_url = 'redis://:@127.0.0.1:6379/1'
from celery import Celery

app = Celery('demo')

app.config_from_object('celeryconfig')


  • 标签云

  • 支付宝扫码支持一下

  • 微信扫码支持一下



基于Nginx+Supervisord+uWSGI+Django1.11.1+Python3.6.5构建

京ICP备20007446号-1 & 豫公网安备 41100202000460号

网站地图 & RSS | Feed