Celery
在程序运行过程中，经常会遇到一些耗时耗资源的任务，为了避免这些任务阻塞主进程的运行，我们会采用多线程或异步任务去处理。比如在Web中需要对新注册的用户发一封激活邮件来验证账户，而发邮件本身是一个IO阻塞式任务，如果把它直接放到应用中，就会阻塞主程序的运行，用户需要等待邮件发送完成后才能进行下一步操作。一种解决办法是为每个发邮件任务新开一个线程去执行，当线程很多时这也不是很好的解决办法；更好的方式是在业务逻辑中触发一个发邮件的异步任务，把待发送的任务都发往任务队列，主程序可以继续往下运行。
Celery是一个强大的分布式任务队列，他可以让任务的执行完全脱离主程序，甚至可以把任务分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。它的架构组成如下图：
可以看到， Celery 主要包含以下几个模块：
-
任务模块
充当任务生产者，包含异步任务和定时任务。其中，异步任务通常在业务逻辑中被触发并发往任务队列，而定时任务由 Celery Beat 进程周期性地将任务发往任务队列。
-
消息中间件 Broker
Broker ，即为任务调度队列，接收任务生产者发来的消息，将任务存入队列。 Celery 本身不提供队列服务，官方推荐使用 RabbitMQ 和 Redis 等。
-
任务执行单元 Worker
充当任务消费者。Worker 是执行任务的处理单元，它实时监控消息队列，获取队列中调度的任务，并执行它。
-
任务结果存储 Backend
Backend 用于存储任务的执行结果，以供查询。同消息中间件一样，存储也可使用 RabbitMQ, Redis 和 MongoDB 等。
异步任务
使用 Celery 实现异步任务主要包含三个步骤：
- 创建一个 Celery 实例
- 启动 Celery Worker
- 应用程序调用异步任务
项目结构：
在init.py文件中创建一个Celery实例：
from celery import Celery
cele = Celery('demo') # 创建 Celery 实例
cele.config_from_object('app.celery_config') # 通过 Celery 实例加载配置模块
Celery配置文件，celery_config.py
# encoding: utf-8
# Author: Timeashore
# celery
BROKER_URL = 'redis://localhost:6379/0' # 使用Redis作为消息代理
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 把任务结果存在了Redis
CELERY_TIMEZONE='Asia/Shanghai' # 指定时区，默认是 UTC
CELERY_TASK_SERIALIZER = 'pickle' # 任务序列化和反序列化使用pickle方案
CELERY_RESULT_SERIALIZER = 'json' # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显
CELERY_ACCEPT_CONTENT = ['json','pickle'] # 指定接受的内容类型
CELERY_IMPORTS = ( # 指定导入的任务模块
# 'app.task1',
# 'app.task2'
'app.email'
)
为了简单起见，对于 Broker 和 Backend ，这里都使用 redis 。
在CELERY_IMPORTS中指定task任务文件。
task任务文件，email.py
import time
from app import cele
@cele.task
def havefun(args):
# do something
# ...
# 模拟执行耗时任务
print u"开始执行celery任务"
time.sleep(10)
print u"celery任务结束"
return args
被@cele装饰的函数都成为可被Celery调度的任务。
在应用程序中触发执行celery异步任务，view.py
from ..email import havefun
@main.route('/test_module', methods=['GET','POST'])
def test_module():
# do something
# ...
# 调用异步任务
havefun.delay([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
# havefun.apply_async([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) # 和delay效果一样
return render_template('fake_celery_task.html')
启动程序
1，在 app 同级目录下，启动 Celery Worker 进程
celery -A app worker --loglevel=info
其中：
- 参数 -A 指定了 Celery 实例的位置，本例是在
__init__.py中， Celery 会自动在该文件中寻找 Celery 对象实例，当然，我们也可以自己指定； - 参数 --loglevel 指定了日志级别，默认为 warning ，也可以使用 -l info 来表示；
在生产环境中，我们通常会使用 Supervisor 来控制 Celery Worker 进程。
启动成功后，控制台会显示如下输出：
2，运行后台任务调用Celery任务结果：
使用 delay() 方法将任务发送到消息中间件（ Broker ）， Celery Worker 进程监控到该任务后，就会进行执行。我们将窗口切换到 Worker 的启动窗口，会看到多了两条日志：
这说明任务已经被调度并执行成功。
另外，我们还可以在Python shell中使用，如果想获取执行后的结果，可以这样做：
>>> from app.email import havefun
>>> havefun.delay([1,2,3,4,5])
<AsyncResult: 651c5beb-983e-4df3-ae0e-aaad2a758239>
>>> result = havefun.delay([1,2,3,4,5])
>>> result.ready() # 使用 ready() 判断任务是否执行完毕
False
>>> result.ready()
True
>>> result.get() # 使用 get() 获取任务结果
[1, 2, 3, 4, 5]
>>>
需要注意，如果在views.py中把函数返回值赋值给一个变量，那么主应用程序也会被阻塞，需要等待异步任务返回的结果。因此，实际使用中，不需要把结果赋值。
在应用程序的views.py中我们使用了delay()和apply_async()两种调用后台任务的方法，两者有什么区别呢？
事实上，delay() 方法封装了 apply_async()，delay函数如下：
def delay(self, *partial_args, **partial_kwargs):
return self.apply_async(partial_args, partial_kwargs)
apply_async()函数原型如下：
def apply_async(self, args=(), kwargs={}, **options):
try:
_apply = self._apply_async
except IndexError: # no tasks for chain, etc to find type
return
# For callbacks: extra args are prepended to the stored args.
if args or kwargs or options:
args, kwargs, options = self._merge(args, kwargs, options)
else:
args, kwargs, options = self.args, self.kwargs, self.options
return _apply(args, kwargs, **options)
apply_async() 支持更多的参数，常用的参数如下：
- countdown ：指定多少秒后执行任务
havefun.apply_async(args=(1,), countdown=15) # 15 秒后执行任务
执行后，多了两条日志文件：
[2018-06-05 16:27:16,592: INFO/MainProcess] Received task: app.email.havefun[d1c4ab89-4fcc-491c-bdd4-b5d0b4faab94]
eta:[2018-06-05 16:27:31.590000+08:00]
表示函数在第16 秒被加入到消息队列，第二行eta表示程序会等到第31 秒才去执行这个任务。
- eta (estimated time of arrival)：指定任务被调度的具体时间，参数类型是 datetime
from datetime import datetime, timedelta
# 当前 UTC 时间再加 10 秒后执行任务
havefun.apply_async(args=(1,), eta=datetime.utcnow() + timedelta(seconds=10))
- expires ：任务过期时间，参数类型可以是 int ，也可以是 datetime
havefun.apply_async(args=(1,), expires=10) # 10 秒后过期
如当前参数设置满足不了需求，请查阅官方文档。
定时任务
Celery 除了可以执行异步任务，也支持执行周期性/定时任务（ Periodic Tasks ）。 Celery Beat 进程通过读取配置文件的内容，周期性地/定时将任务发往任务队列。
同样的，我们还使用上述的目录结构；不同的是，我们只需要在Celery配置文件 celery_config.py 文件中添加配置 CELERYBEAT_SCHEDULE 字段并在 CELERY_IMPORTS 中导入任务模块就可以实现定时任务，具体如下：
# encoding: utf-8
# Author: Timeashore
from datetime import timedelta
from celery.schedules import crontab
# celery
BROKER_URL = 'redis://localhost:6379/0' # 使用Redis作为消息代理
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # 把任务结果存在了Redis
CELERY_TIMEZONE='Asia/Shanghai' # 指定时区，默认是 UTC
CELERY_TASK_SERIALIZER = 'pickle' # 任务序列化和反序列化使用pickle方案
CELERY_RESULT_SERIALIZER = 'json' # 读取任务结果一般性能要求不高，所以使用了可读性更好的JSON
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24 # 任务过期时间，不建议直接写86400，应该让这样的magic数字表述更明显
CELERY_ACCEPT_CONTENT = ['json','pickle'] # 指定接受的内容类型
CELERY_IMPORTS = ( # 指定导入的任务模块
'app.task1',
'app.task2'
'app.email'
)
# schedules
CELERYBEAT_SCHEDULE = {
'add-every-30-seconds': {
'task': 'app.task1.add',
'schedule': timedelta(seconds=30), # 每 30 秒一次
# 'schedule': timedelta(minutes=1), # 每 1 分钟一次
# 'schedule': timedelta(hours=4), # 每 4 小时一次
'args': (5, 8) # 任务函数参数
},
'multiply-at-some-time': {
'task': 'app.task2.multiply',
'schedule': crontab(hour=9, minute=50), # 每天早上 9 点 50 分执行一次
'args': (3, 7) # 任务函数参数
}
}
我们还需在 task 任务 email.py 同级目录下添加两个任务文件 task1.py , task2.py 。
task1.py
import time
from app import cele
@cele.task
def add(x, y):
time.sleep(2)
return x + y
task2.py
import time
from app import cele
@cele.task
def multiply(x, y):
time.sleep(2)
return x * y
运行
关闭上次的程序，重新启动 Celery Worker 进程
celery -A app worker --loglevel=info
然后，app同级目录，启动 Celery Beat 进程，定时将任务发送到 Broker
celery beat -A app
手动调用 2 次 Celery 任务, havefun()：
查看日志文件，在此为了好展示，仅添加了 task1.py 文件并设置 1 分钟执行一次：
Celery 周期性任务也有多个配置项，如当前设置满足不了需求，请查阅官方文档。
END！
