Django 2.1.7 集成Celery 4.3.0 从介绍到入门

Django 2.1.7 集成Celery 4.3.0 从介绍到入门

相关篇章：​​Django 2.1.7 Celery 4.3.0 示例，解决Task handler raised error: ValueError('not enough values to unp...​​

问题抛出

我们在做网站后端程序开发时，会碰到这样的需求：用户需要在我们的网站填写注册信息，我们发给用户一封注册激活邮件到用户邮箱，如果由于各种原因，这封邮件发送所需时间较长，那么客户端将会等待很久，造成不好的用户体验.

那么怎么解决这样的问题呢?

我们将耗时任务放到后台异步执行。不会影响用户其他操作。除了注册功能，例如上传，图形处理等等耗时的任务，都可以按照这种思路来解决。 如何实现异步执行任务呢？我们可使用celery. celery除了刚才所涉及到的异步执行任务之外，还可以实现定时处理某些任务。

celery介绍

Celery是一个功能完备即插即用的任务队列。它使得我们不需要考虑复杂的问题，使用非常简单。celery看起来似乎很庞大，我们先对其进行简单的了解，然后再去学习其他一些高级特性。 celery适用异步处理问题，当发送邮件、或者文件上传, 图像处理等等一些比较耗时的操作，我们可将其异步执行，这样用户不需要等待很久，提高用户体验。 celery的特点是：

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

celery非常易于集成到一些web开发框架中.

下面来看看一些概念。

Task Queue 任务队列

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

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

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

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

Celery安装

我们可以使用python的包管理器pip来安装:

pip install -U Celery

也可从官方直接-安装包:​​xvfz celery-0.0.0.tar.gzcd celery-0.0.0python setup.py buildpython setup.py install

除了安装Celery，因为本次示例需要在django 2.1.7中使用。所以还要安装其他库。

所有库安装命令：

# 安装最新版本celerypip3 install -U Celery# 安装django集成celerypip3 install django-celery==3.3.0# 安装celery的相关依赖库，用于连接redis、rabbitmq等等pip3 install "celery[librabbitmq,redis,auth,msgpack]"# 解决win10启动worker问题pip3 install eventlet

版本信息

python 3.7.2 或者 3.7.1 （当前我使用这两个版本测试）celery 4.3.0django-celery 3.3.0

Borker 中间人代理

Celery需要一种解决消息的发送和接受的方式，我们把这种用来存储消息的的中间装置叫做message broker, 也可叫做消息中间人。 作为中间人，我们有几种方案可选择：

1.RabbitMQ

RabbitMQ是一个功能完备，稳定的并且易于安装的broker. 它是生产环境中最优的选择。使用RabbitMQ的细节参照以下链接： ​​sudo apt-get install rabbitmq-server 安装完毕之后，RabbitMQ-server服务器就已经在后台运行。如果您用的并不是Ubuntu或Debian, 可以在以下网址： ​​去查找自己所需要的版本软件。

2.Redis

Redis也是一款功能完备的broker可选项，但是其更可能因意外中断或者电源故障导致数据丢失的情况。 关于是有那个Redis作为Broker，可访下面网址： ​​应用

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

1.创建应用

首先在Django项目中创建一个celery_tasks文件夹，再创建tasks.py模块, 如下：

编写tasks.py 其内容为:

from celery import Celery# 使用redis作为brokerapp = Celery('celery_tasks.tasks', broker='redis://127.0.0.1:6379/8')# 创建任务函数@app.taskdef my_task(): print("任务函数正在执行....")

Celery第一个参数是给其设定一个名字， 第二参数我们设定一个中间人broker, 在这里我们使用Redis作为中间人。my_task函数是我们编写的一个任务函数， 通过加上装饰器app.task, 将其注册到broker的队列中。

现在我们在创建一个worker， 等待处理队列中的任务.打开终端，cd到tasks.py同级目录中，执行命令:​​​celery -A celery_tasks.tasks worker -l info -P eventlet​​

启动过程如下：

(venv) F:\pythonProject\django-pratice>celery -A celery_tasks.tasks worker -l info -P eventlet -------------- celery@USC2VG2F9NPB650 v4.3.0 (rhubarb)---- **** -------- * *** * -- Windows-10-10.0.17763-SP0 2019-08-03 00:33:17-- * - **** ---- ** ---------- [config]- ** ---------- .> app: celery_tasks.tasks:0x278a66264a8- ** ---------- .> transport: redis://127.0.0.1:6379/8- ** ---------- .> results: disabled://- *** --- * --- .> concurrency: 12 (eventlet)-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)--- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery[tasks] . celery_tasks.tasks.my_task[2019-08-03 00:33:17,385: INFO/MainProcess] Connected to redis://127.0.0.1:6379/8[2019-08-03 00:33:17,425: INFO/MainProcess] mingle: searching for neighbors[2019-08-03 00:33:18,665: INFO/MainProcess] mingle: all alone[2019-08-03 00:33:18,744: INFO/MainProcess] pidbox: Connected to redis://127.0.0.1:6379/8.[2019-08-03 00:33:18,805: INFO/MainProcess] celery@USC2VG2F9NPB650 ready.

2.调用任务

任务加入到broker队列中，以便刚才我们创建的celery workder服务器能够从队列中取出任务并执行。如何将任务函数加入到队列中，可使用delay()。

进入python终端, 执行如下代码:

In [3]: from celery_tasks.tasks import my_task# 调用一个任务函数，将会返回一个AsyncResult对象，这个对象可以用来检查任务的状态或者获得任务的返回值。In [4]: my_task.delay()Out[4]:

返回worker的终端界面，查看任务执行情况，如下：

可以看到已经收到任务，并执行打印了信息。

3.存储结果

如果我们想跟踪任务的状态，Celery需要将结果保存到某个地方。有几种保存的方案可选:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

例子我们仍然使用Redis作为存储结果的方案，任务结果存储配置我们通过Celery的backend参数来设定。我们将tasks模块修改如下:

from celery import Celeryapp = Celery('celery_tasks.tasks', broker='redis://127.0.0.1:6379/8', backend='redis://127.0.0.1:6379/9',)@app.taskdef my_task(a, b): print("任务函数正在执行....") return a + b

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

下面再来执行调用一下这个任务看看。

In [1]: from celery_tasks.tasks import my_task# 传递参数至任务中In [5]: ret = my_task.delay(10,20)# 查询返回值的结果In [6]: ret.resultOut[6]: 30# 查看是否执行失败In [7]: ret.failed()Out[7]: False

再来看看worker的执行情况，如下：

可以看到celery任务已经执行成功了。

要注意的是如果修改了任务代码，那么就需要重启celery的worker重新注册，不然不会生效。

更多关于result对象信息，请参阅下列网址:​​http://docs.celeryproject.org/en/latest/reference/celery.result.html#module-celery.result​​

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。