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
复制代码

也可从官方直接下载安装包:pypi.python.org/pypi/celery…

tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python setup.py build
python setup.py install
复制代码

除了安装Celery，由于本次示例须要在django 2.1.7中使用。因此还要安装其余库。

全部库安装命令：

# 安装最新版本celery
pip3 install -U Celery

# 安装django集成celery
pip3 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.0
• django-celery 3.3.0

Borker 中间人代理

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

1.RabbitMQ

RabbitMQ是一个功能完备，稳定的而且易于安装的broker. 它是生产环境中最优的选择。使用RabbitMQ的细节参照如下连接： docs.celeryproject.org/en/latest/g…

若是咱们使用的是Ubuntu或者Debian发行版的Linux，能够直接经过下面的命令安装RabbitMQ: sudo apt-get install rabbitmq-server 安装完毕以后，RabbitMQ-server服务器就已经在后台运行。若是您用的并非Ubuntu或Debian, 能够在如下网址： www.rabbitmq.com/download.ht… 去查找本身所须要的版本软件。

2.Redis

Redis也是一款功能完备的broker可选项，可是其更可能因意外中断或者电源故障致使数据丢失的状况。 关因而有那个Redis做为Broker，可访下面网址： docs.celeryproject.org/en/latest/g…

Application 应用

使用celery第一件要作的最为重要的事情是须要先建立一个Celery实例，咱们通常叫作celery应用，或者更简单直接叫作一个app。app应用是咱们使用celery全部功能的入口，好比建立任务，管理任务等，在使用celery的时候，app必须可以被其余的模块导入。

1.建立应用

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

编写tasks.py 其内容为:

from celery import Celery

# 使用redis做为broker
app = Celery('celery_tasks.tasks', broker='redis://127.0.0.1:6379/8')

# 建立任务函数
@app.task
def 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]: <AsyncResult: 647b2589-95d2-45c9-a9a7-0b5530caf249>
复制代码

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

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

3.存储结果

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

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

from celery import Celery

app = Celery('celery_tasks.tasks',
             broker='redis://127.0.0.1:6379/8',
             backend='redis://127.0.0.1:6379/9',)

@app.task
def 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.result
Out[6]: 30

# 查看是否执行失败
In [7]: ret.failed()
Out[7]: False

复制代码

再来看看worker的执行状况，以下：

能够看到celery任务已经执行成功了。

要注意的是若是修改了任务代码，那么就须要重启celery的worker从新注册，否则不会生效。

更多关于result对象信息，请参阅下列网址:docs.celeryproject.org/en/latest/r…