Celery简介
Celery(芹菜)是基于Python开发的分布式任务队列。它支持使用任务队列的方式在分布的机器/进程/线程上执行任务调度。
Celery架构
架构图如下:
Celery包括如下组件:
- Celery Beat
任务调度器,Beat进程会读取配置文件的内容,周期性的将配置中到期需要执行的任务发送给任务队列
- celery Worker
执行任务的消费者,通常会在多台服务器运行多个消费者来提高执行效率
- Broker
消息代理,或者叫做消息中间件,接受任务生产者发送过来的任务消息,存进队列在按序分发给任务消费方(通常是消息队列或者数据库)
- Producer
调用了Celery提供的API,函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者
- Result Backend
任务处理完后保存状态信息和结果,以供查询。Celery默认已支持 Redis,RabbitMQ, MongoDB,Django ORM,SQLAIchemy等方式
中间件选择
Celery目前支持很多第三方软件作为消息代理,但适用于生产环境的只有RabbitMQ和Redis,至于其他的方式,一是支持有限,二是可能得不到更好的技术支持。Celery官方推荐的是RabbitMQ。
Celery序列化
在客户端和消费者之间传输数据需要序列化和反序列化,Celery支持如下表的序列化方案:
方案 | 说明 |
---|---|
pickle | pickle是Python标准库中的一个模块,支持Python内置的数据结构,但是它是Python的专有协议。 从Celery3.2开始,由于安全性等原因Celery将拒绝pickle这个方案 |
json | json支持多种语言,可用于跨语言方案 |
yaml | yaml的表达能力更强,支持的数据类型比json多,但是python客户端的性能不如json |
msgpack | msgpack是一个二进制的类json的序列化方案,但是比json的数据结构更小、更快 |
简单项目
项目目录结构如下:
/root/test/proj/celery
├── celeryconfig.py
├── celery.py
├── __init__.py
└── tasks.py
先看一下主程序celery.py:
#!/usr/bin/env python
#coding:utf8
#拒绝隐式引入,因为celery.py的名字和celery的包名冲突,需要使用这条语句让程序正常运行,否则“from celery import Celery”这条语句将会报错,因为首先找到的celery.py文件中并没有Celery这个类
from __future__ import absolute_import
from celery import Celery
# app是Celery类的实例,创建的时候添加了celery.tasks这个模块,也就是包含了celery/tasks.py这个文件
app = Celery('celery',include=['celery.tasks'])
# 把Celery配置存放进celery/celeryconfig.py文件,使用app.config_from_object加载配置
app.config_from_object('celery.celeryconfig')
if __name__ == "__main__":
app.start()
存放任务函数的文件tasks.py:
#!/usr/bin/env python
#coding:utf8
from __future__ import absolute_import
from celery.celery import app
@app.task
def add(x, y):
return x+y
tasks.py只有一个任务函数add,让它生效的最直接的方法就是添加app.task这个装饰器。
celery配置文件celeryconfig.py:
# 使用Redis作为消息代理
BROKER_URL = 'redis://192.168.189.100:6379/0'
# 把任务结果保存在Redis中
CELERY_RESULT_BACKEND = 'redis://192.168.189.100:6379/1'
# 任务序列化和反序列化使用msgpack方案
CELERY_TASK_SERIALIZER = 'msgpack'
# 读取任务结果一般性能要求不高,所以使用了可读性更好的JSON
CELERY_RESULT_SERIALIZER = 'json'
# 任务过期时间,这样写更加明显
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 指定接受的内容类型
CELERY_ACCEPT_CONTENT = ['json', 'msgpack']
这个例子中没有任务调度相关的内容,如果有的话就要使用Queue类了,所以只需要启动消费者:
celery -A proj worker -l info
-A参数默认会寻找proj.celery这个模块,其实使用celery作为模块文件名字不怎么合理。可以使用其他名字。举个例子,假如是proj/app.py,可以使用如下命令启动:
celery -A proj.app worker -l info