Celery从入门到出家

在学习Celery之前，先简单的去了解了一下什么是生产者消费者模式。

生产者消费者模式

在实际的软件开发过程中，经常会碰到如下场景：某个模块负责产生数据，这些数据由另一个模块来负责处理（此处的模块是广义的，可以是类、函数、线程、进程等）。产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者。

单单抽象出生产者和消费者，还够不上是生产者消费者模式。该模式还需要有一个缓冲区处于生产者和消费者之间，作为一个中介。生产者把数据放入缓冲区，而消费者从缓冲区取出数据，如下图所示：

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过消息队列（缓冲区）来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给消息队列，消费者不找生产者要数据，而是直接从消息队列里取，消息队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个消息队列就是用来给生产者和消费者解耦的。------------->这里又有一个问题，什么叫做解耦？
解耦：假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化，可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。生产者直接调用消费者的某个方法，还有另一个弊端。由于函数调用是同步的（或者叫阻塞的），在消费者的方法没有返回之前，生产者只好一直等在那边。万一消费者处理数据很慢，生产者就会白白糟蹋大好时光。缓冲区还有另一个好处。如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来，消费者再慢慢处理掉。

因为太抽象，看过网上的说明之后，通过我的理解，我举了个例子：吃包子。

假如你非常喜欢吃包子（吃起来根本停不下来），今天，你妈妈（生产者）在蒸包子，厨房有张桌子（缓冲区），你妈妈将蒸熟的包子盛在盘子（消息）里，然后放到桌子上，你正在看巴西奥运会，看到蒸熟的包子放在厨房桌子上的盘子里，你就把盘子取走，一边吃包子一边看奥运。在这个过程中，你和你妈妈使用同一个桌子放置盘子和取走盘子，这里桌子就是一个共享对象。生产者添加食物，消费者取走食物。桌子的好处是，你妈妈不用直接把盘子给你，只是负责把包子装在盘子里放到桌子上，如果桌子满了，就不再放了，等待。而且生产者还有其他事情要做，消费者吃包子比较慢，生产者不能一直等消费者吃完包子把盘子放回去再去生产，因为吃包子的人有很多，如果这期间你好朋友来了，和你一起吃包子，生产者不用关注是哪个消费者去桌子上拿盘子，而消费者只去关注桌子上有没有放盘子，如果有，就端过来吃盘子中的包子，没有的话就等待。对应关系如下图：

考察了一下，原来当初设计这个模式，主要就是用来处理并发问题的，而Celery就是一个用python写的并行分布式框架。

Celery的定义

Celery（芹菜）是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。

我比较喜欢的一点是：Celery支持使用任务队列的方式在分布的机器、进程、线程上执行任务调度。

而分布式的异步任务队列,让应用程序可能需要执行任何消耗资源的任务都交给任务队列,让应用程序能够自如快速地相应客户端地请求

任务队列

任务队列是一种被用来向线程或者机器分发任务的机制,一个任务队列输入的单元被称为一个task,专用的worker线程持续的监听任务队列等待新的任务出现去执行.

Celery的通信通过消息来执行,通常使用一个broker来在客户端和worker之间作为中间件.初始化一个任务时,客户端发送一个message给任务队列,然后broker分发message向各个worker

Celery需要一个消息缓存区发送和接受消息,RabbitMQ和Redis来作为broker,即消息的中间件

异步任务: 简而言之，做一个注册的功能，在用户使用邮箱注册成功之后，需要给该邮箱发送一封激活邮件。如果直接放在应用中，则调用发邮件的过程会遇到网络IO的阻塞，比好优雅的方式则是使用异步任务，应用在业务逻辑中触发一个异步任务。

消息队列

消息队列的输入是工作的一个单元，称为任务，独立的职程（Worker）进程持续监视队列中是否有需要处理的新任务。

Celery 用消息通信，通常使用中间人（Broker）在客户端和职程间斡旋。这个过程从客户端向队列添加消息开始，之后中间人把消息派送给职程，职程对消息进行处理。如下图所示：

Celery的架构

Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。
Celery 是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。
我们通常使用它来实现异步任务（async task）和定时任务（crontab）
我用一张图给大家展示看一下：

Celery Beat：任务调度器，Beat进程会读取配置文件的内容，周期性地将配置中到期需要执行的任务发送给任务队列。
Celery Worker：执行任务的消费者，通常会在多台服务器运行多个消费者来提高执行效率。
Broker：消息代理，或者叫作消息中间件，接受任务生产者发送过来的任务消息，存进队列再按序分发给任务消费方（Celery目前支持RabbitMQ、Redis、MongoDB、Beanstalk、SQLAlchemy、Zookeeper等作为消息代理，但适用于生产环境的只有RabbitMQ和Redis）。
Producer：调用了Celery提供的API、函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。
Result Backend：任务处理完后保存状态信息和结果，以供查询。Celery默认已支持Redis、RabbitMQ、MongoDB、Django ORM、SQLAlchemy等方式。

消息中间件（message broker）

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成，包括，RabbitMQ,Redis,MongoDB等。

任务执行单元（worker）

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。

任务执行结果存储（task result store）

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括Redis，MongoDB，Django ORM，AMQP等，这里我先不去看它是如何存储的，就先选用Redis来存储任务执行结果。

特性

Monitoring: 被分发给worker的事件,可以实时的知道哪一个聚簇在工作
Scheduing: 指定特定的时间去执行某个任务,或者定期反复的去执行某个任务
Work-flows: 简单和复杂的工作流通过canvas组合在一起?
Resource Leak Protection: –max-tasks-per-child 用于资源泄露保护
Time &Rate Limits:可以控制每秒/分/时,有多少个任务可以被执行,或者说一个任务被允许执行多长的时间
理解了celery之后，下面我们直接上代码（安装我就不说了，自己研究）

celery实现

celery异步任务和调用执行任务

# tasks.py
from celery import Celery
app = Celery('tasks',broker = 'redis://password@127.0.0.1:6379/5',backend = 'redis://password@127.0.0.1:6379/6'
)@app.task
def add(x, y):return x + y

到这里相信很多朋友对参数有些懵，我这里解释下Celery实例化对象的参数：

tasks	为当前模块的名称,这个用来自动的获取在 main\ module中的任务
main	如果作为__main__运行，则为主模块的名称。用作自动生成的任务名称的前缀
loader	当前加载器实例
backend	任务结果url
amqp	AMQP对象或类名，一般不管
log	日志对象或类名
set_as_current	将本实例设为全局当前应用
tasks	任务注册表。
broker	使用的默认代理的URL,任务队列；
include	每个worker应该导入的模块列表，以实例创建的模块的目录作为起始路径

运行启动celery命令：celery -A tasks worker -l info
加入队列启动任务命令：celery -A tasks worker -Q queue -l info

查询文档，了解到该命令中-A参数表示的是Celery APP的名称，这个实例中指的就是tasks.py,后面的tasks就是APP的名称，worker是一个执行任务角色，后面的loglevel=info记录日志类型默认是info,这个命令启动了一个worker,用来执行程序中add这个加法任务（task）。

关于执行参数的解释，参数太多了我就不一一解释了，大家可以 help 查看下

然后看到界面显示结果如下：

我们可以看到Celery 正常工作在名称ubuntu的虚拟主机上，版本为3.1.23，在下面的 [config]中我们可以看到当前APP的名称tasks，运输工具transport就是我们在程序中设置的中间人（broker）redis://127.0.0.1:6379/5，result（backend）我们没有设置，暂时显示为disabled（）,然后我们也可以看到worker缺省使用perfork来执行并发，当前并发数显示为1，然后可以看到下面的**[queues]就是我们说的队列**，当前默认的队列是celery,然后我们看到下面的**[tasks]中有一个任务tasks.add**.

实例化对象参数

任何被 task 修饰的方法都会被创建一个 Task 对象，变成一个可序列化并发送到远程服务器的任务；它有多种修饰方式：

方式一：使用默认的参数

@celery.task
def function_name():pass

方式二：指定相关参数

@celery.task(bind=True, name='name')
def function_name():pass# task方法参数
name       : 可以显式指定任务的名字；默认是模块的命名空间中本函数的名字。
serializer : 指定本任务的序列化的方法；
bind       : 一个bool值，设置是否绑定一个task的实例，如果绑定，task实例会作为参数传递到任务方法中，可以访问task实例的所有的属性，即前面反序列化中那些属性
base       : 定义任务的基类，可以以此来定义回调函数，默认是Task类，我们也可以定义自己的Task类
default_retry_delay : 设置该任务重试的延迟时间，当任务执行失败后，会自动重试，单位是秒，默认3分钟；
autoretry_for       : 设置在特定异常时重试任务，默认False即不重试；
retry_backoff       : 默认False，设置重试时的延迟时间间隔策略；
retry_backoff_max   : 设置最大延迟重试时间，默认10分钟，如果失败则不再重试；
retry_jitter        : 默认True，即引入抖动，避免重试任务集中执行；# 当bind=True时，add函数第一个参数是self，指的是task实例
@task(bind=True)  # 第一个参数是self，使用self.request访问相关的属性
def add(self, x, y):try:logger.info(self.request.id)except:self.retry() # 当任务失败则进行重试，也可以通过max_retries属性来指定最大重试次数

方式三：自定义Task基类

import celeryclass MyTask(celery.Task):# 任务失败时执行def on_failure(self, exc, task_id, args, kwargs, einfo):print('{0!r} failed: {1!r}'.format(task_id, exc))# 任务成功时执行def on_success(self, retval, task_id, args, kwargs):pass# 任务重试时执行def on_retry(self, exc, task_id, args, kwargs, einfo):pass@task(base=MyTask)
def add(x, y):raise KeyError()# 方法相关的参数
exc     : 失败时的错误的类型；
task_id : 任务的id；
args    : 任务函数的参数；
kwargs  : 键值对参数；
einfo   : 失败或重试时的异常详细信息；
retval  : 任务成功执行的返回值；

Task的常用属性

Task.name     : 任务名称；
Task.request  : 当前任务的信息；
Task.max_retries   : 设置重试的最大次数
Task.throws        : 预期错误类的可选元组，不应被视为实际错误，而是结果失败；
Task.rate_limit    : 设置此任务类型的速率限制
Task.time_limit    : 此任务的硬限时（以秒为单位）。
Task.ignore_result : 不存储任务状态。默认False；
Task.store_errors_even_if_ignored : 如果True，即使任务配置为忽略结果，也会存储错误。
Task.serializer    : 标识要使用的默认序列化方法的字符串。
Task.compression   : 标识要使用的默认压缩方案的字符串。默认为task_compression设置。
Task.backend       : 指定该任务的结果存储后端用于此任务。
Task.acks_late     : 如果设置True为此任务的消息将在任务执行后确认 ，而不是在执行任务之前（默认行为），即默认任务执行之前就会发送确认；
Task.track_started : 如果True任务在工作人员执行任务时将其状态报告为“已启动”。默认是False；

了解了这些之后，我们可以调用一下任务

调用异步任务的三种方法

调用异步任务的三个方法分别是：

方法一：这是apply_async方法的别名,但接受的参数较为简单；
task.delay()
delay 方法是 apply_async 方法的简化版，不支持执行选项，只能传递任务的参数。
```
from celery import Celeryapp = Celery()@app.task
def add(x, y, z=0):return x + yadd.delay(30, 40, z=5)    # 包括位置参数和关键字参数
```

方法二：可以接受复杂的参数
task.apply_async(args=[arg1, arg2], kwargs={key:value, key:value})
详细参数见官方文档
apply_async 支持执行选项，它会覆盖全局的默认参数和定义该任务时指定的执行选项，本质上还是调用了 send_task 方法；

from celery import Celeryapp = Celery()@app.task
def add(x, y, z=0):return x + yadd.apply_async(args=[30,40], kwargs={'z':5})
# 其他参数
task_id   : 为任务分配唯一id，默认是uuid;
countdown : 设置该任务等待一段时间再执行，单位为s；
eta       : 定义任务的开始时间；eta=time.time()+10;
expires   : 设置任务时间，任务在过期时间后还没有执行则被丢弃；
retry     : 如果任务失败后, 是否重试;使用true或false，默认为true
shadow    : 重新指定任务的名字str，覆盖其在日志中使用的任务名称；
retry_policy : {},重试策略.如下：
----max_retries    : 最大重试次数, 默认为 3 次.
----interval_start : 重试等待的时间间隔秒数, 默认为 0 , 表示直接重试不等待.
----interval_step  : 每次重试让重试间隔增加的秒数, 可以是数字或浮点数, 默认为 0.2
----interval_max   : 重试间隔最大的秒数, 即 通过 interval_step 增大到多少秒之后, 就不在增加了, 可以是数字或者浮点数, 默认为 0.2 .routing_key : 自定义路由键；
queue       : 指定发送到哪个队列；
exchange    : 指定发送到哪个交换机；
priority    : 任务队列的优先级，0到255之间，对于rabbitmq来说0是最高优先级；
serializer  :任务序列化方法；通常不设置；
compression : 压缩方案，通常有zlib, bzip2
headers     : 为任务添加额外的消息；
link        : 任务成功执行后的回调方法；是一个signature对象；可以用作关联任务；
link_error  : 任务失败后的回调方法，是一个signature对象；# 其他参数参考用法如下：
add.apply_async((2, 2), retry=True, retry_policy={'max_retries': 3,
'interval_start': 0,
'interval_step': 0.2,
'interval_max': 0.2,
})

重要参数(parameters):

args(Tuple) :传送给任务的位置参数
kwargs(Dict) :传送给任务的关键字参数
countdown(float) :在多少秒之后,任务将会被执行,默认为立即执行
expires(float,datetime) :任务将会在未来多少秒或时间后被终结,任务将不会在终结时间之后执行
connection(kombu.Connection): 重复使用的中间件链接而不是每次都去从链接池中请求
retry(bool) :失败连接retry ==>setting
retry_policy(Mapping) ==>setting

方法三：可以发送未被注册的异步任务，即没有被celery.task装饰的任务；
send_task()
而 send_task 方法最为特殊，需要使用实例化对象 app 来直接调用执行，因为他的好处就是可以发送不需要注册的任务
app.send_task()
注意： send_task 在发送的时候是不会检查 tasks.add 函数是否存在的，即使为空也会发送成功，所以 celery 执行是可能找不到该函数报错；
```
# File_name：tasks.py
from celery import Celeryapp = Celery()def add(x, y):return x+yapp.send_task('tasks.add',args=[3,4])  # 参数基本和apply_async函数一样，在这里我解释下参数，第一个位置的 tasks.add ，tasks是文件名，add是函数名，后面的参数就和apply_async是一样的了！
```

了解了这些之后，根据文档我重新打开一个terminal,然后执行Python,进入Python交互界面，用delay()方法调用任务，执行如下操作：

这个任务已经由之前启动的Worker异步执行了，然后我打开之前启动的worker的控制台,对输出进行查看验证，结果如下：

绿色部分第一行说明worker收到了一个任务：tasks.add,这里我们和之前发送任务返回的AsyncResult对比我们发现，每个task都有一个唯一的ID，第二行说明了这个任务执行succeed,执行结果为12。
查看资料说调用任务后会返回一个AsyncResult实例，可用于检查任务的状态，等待任务完成或获取返回值（如果任务失败，则为异常和回溯）。但这个功能默认是不开启的，需要设置一个 Celery 的结果后端（backend）
另外两种方法可以自己写一个小deam，这里就不一一展示了
通过这个例子后我对Celery有了初步的了解，然后我在这个例子的基础上去进一步的学习。

celery定时任务

上面我讲解和展示了一些异步任务的实现和调用，相信大家也对celery有了认识和了解。下面我来说一下celery重要功能之一的定时任务

Celery加入定时任务

Celery除了可以异步执行任务之外，还可以定时执行任务。在实例代码的基础上写个测试方法：

# -*- coding : utf-8 -*-
from __future__ import absolute_import
from celery import Celery
from celery.decorators import periodic_task
from celery.schedules import crontabapp = Celery('celeryapp',broker='redis://password@127.0.0.1:6379/3',backend='redis://password@127.0.0.1:6379/4',
)@periodic_task(run_every=crontab(minute='*'))
def one():print("--------------------------------------one is running...")return True@periodic_task(run_every=crontab(minute='*/3'))
def two():print("--------------------------------------two is running...")return True

代码是每分钟执行一次
启动celery是使用worker。但worker不能启动定时任务。启动方式如下：

celery -A celeryapp beat -l info

这个beat是检查定时任务，并启动定时任务丢给worker执行。如下图，可以看到一点，beat启动定时任务之后，两个定时任务会立马先执行一次，之后才会按设置的时间执行定时任务

等一分钟后可看到下图
beat窗口

worker窗口

这样，定时任务就起来啦！很简单吧！大家多些几个deam练习一下就会很熟了。

Celery定时任务时间设置

若你觉得1分钟等待时间太长。可以设置为每10秒执行一次定时任务。将上面的代码修改如下：

@periodic_task(run_every=10)

修改代码，需要重启Celery的worker和beat。
这个run_every参数是设置定时任务的时间间隔或执行的时间。该参数设置有3种方式。

直接设置秒数
例如刚刚所说的10秒间隔，run_every=10，每10秒执行一次任务。1分钟即是60秒；1小时即是3600秒。
通过datetime设置时间间隔
有时直接设置秒数不方便，需要通过计算得到具体秒数。

例如，1小时15分钟40秒 = 16060 + 15*60 + 40。这种情况可读性也不高。

可以采用datetime设置，代码如下：
```
from celery.decorators import periodic_task
import datetime@periodic_task(run_every=datetime.timedelta(hours=1, minutes=15, seconds=40))
def one():print("--------------------------------------one is running...")return True
```
代码可读性明显提升，而且设置方便。

但这种不能满足定时定点的时间设置。假如我想固定每天12点15分的时候，执行一次任务。datetime和直接设置秒数的方式都无法实现。这时得使用第3种方式。

celery的crontab表达式
crontab是比较完善，且稍微有点复杂（相对前面两种方式而言）的方式。可以实现我们各种设置时间的需求。

# 上面就是用的这种方式，细心的朋友肯定也已经发现了
@periodic_task(run_every=crontab(minute='*'))

表示每分钟0秒时刻执行一次（后面不提这个0秒，大家都知道就行了，省点口水）。
其中，crontab()实例化的时候没设置任何参数，都是使用默认值。crontab一共有7个参数，常用有5个参数分别为：

minute：分钟，范围0-59；hour：小时，范围0-23；day_of_week：星期几，范围0-6。以星期天为开始，即0为星期天。这个星期几还可以使用英文缩写表示，例如“sun”表示星期天；day_of_month：每月第几号，范围1-31；month_of_year：月份，范围1-12。

默认参数

这些参数可以设置表达式，表达稍微复杂的设置。默认值都是"*"星号，代表任意时刻。即crontab()相当与：

crontab(minute='*', hour='*', day_of_week='*', day_of_month='*', month_of_year='*')

含义是每天、每小时、每分钟执行一次任务。这说法太反人类语言习惯，简单说就是每1分钟执行一次任务。

具体某个值

上面提到这些参数的取值范围。我们可以直接设置某个值。例如：

crontab(minute=15)

即每小时的15分时刻执行一次任务。直接指定某个时刻。以此类推可以设置每天0点0分时刻执行任务的设置如下：

crontab(minute=0, hour=0)

当然，也可以设置多个值。例如0分和30分执行一次任务：

crontab(minute='0,30')

这里使用字符串，用逗号隔开数值。这里的逗号是表示多个表达式or逻辑关系。

设置范围

设置范围也是设置多个值，例如指定9点到12点每个小时的每分钟执行任务。

crontab(minute='*', hour='9-12')

这里*号是默认值，可以省略如下：

crontab(hour='9-12')

上面提到逗号是or逻辑关系。拓展一下，指定9点到12点和20点中每分钟执行任务：

crontab(hour='9-12,20')

crontab的表达式越来越复杂了。celery还提供了一个类得到表达式解析结果，代码如下：

from celery.schedules import crontab_parser
r = crontab_parser(23, 0).parse('9-12,20')
print(r)

其中，crontab_parse是一个解析类。第1个参数是范围的最大值；第2个参数是范围的最小值。通过parse输入表达式，可得到表达式的解析结果：

set([9, 10, 11, 12, 20])

下面很多地方我们都可以通过该方法验证解析结果。

设置间隔步长

假如我要设置1、3、5、7、9、11月份每天每分钟执行任务，按照上面的做法可以设置如下：

crontab(day_of_month='1,3,5,7,9,11')

 观察数据可以发现，都是间隔2的步长。需要设置的数字比较少，若数字比较多显得很麻烦。例如我想每间隔2分钟就执行一次任务，要写30个数字想想就觉得很麻烦。crontab表达式还提供了间隔的处理，例如：

crontab(minute='*/2')
crontab(minute='0-59/2') #效果等同上面

 这个/号不是除以的意思。相当与range的第3个参数，例如：

range(0, 59+1, 2)

这里举两个例子，大家就都明白了

每分钟执行一次
crontab() 或者 crontab(minute='*')每天凌晨十二点执行
crontab(minute=0, hour=0)每十五分钟执行一次
crontab(minute=’*/15’)每周日的每一分钟执行一次
crontab(minute=’’,hour=’’, day_of_week=‘sun’)每周三，五的三点，七点和二十二点没十分钟执行一次
crontab(minute=’*/10’,hour=‘3,17,22’, day_of_week=‘thu,fri’)#每2个小时中每分钟执行1次任务
crontab(hour='*/2')#每3个小时的0分时刻执行1次任务
#即[0,3,6,9,12,15,18,21]点0分
crontab(minute=0, hour='*/3')#每3个小时或8点到12点的0分时刻执行1次任务
#即[0,3,6,9,12,15,18,21]+[8,9,10,11,12]点0分
crontab(minute=0, hour='*/3,8-12')#每个季度的第1个月中，每天每分钟执行1次任务
#月份范围是1-12，每3个月为[1,4,7,10]
crontab(month_of_year='*/3')#每月偶数天数的0点0分时刻执行1次任务
crontab(minute=0, hour=0, day_of_month='2-31/2')#每年5月11号的0点0分时刻执行1次任务
crontab(0, 0, day_of_month='11', month_of_year='5')

更多请参见：官方文档
好了，celery就先说到这里了，大家有什么想法或者发现了更好玩的东西，欢迎大家到评论区或者私信我大家一起讨论下哦！