Celery是一个简单,灵活且能处理异步任务,定时任务及大量消息的分布式系统
专注于实时处理的异步任务队列
同时也支持任务调度
Celery框架由三部分组成:消息中间件(AMQP broker),任务执行单元(celery workers),任务执行结果存储(task result store)组成
消息中间件
celery本身不提供消息服务,但是方便和第三方提供的消息中间件集成,包括RabbitMQ,redis等
任务执行单元
Worker是celery提供的任务执行单元worker并发的运行在任务节点中
任务执行结果存储
Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等
版本支持情况
Celery version 4.0 runs on Python ❨2.7, 3.4, 3.5❩ PyPy ❨5.4, 5.5❩ This is the last version to support Python 2.7, and from the next version (Celery 5.x) Python 3.5 or newer is required. If you’re running an older version of Python, you need to be running an older version of Celery: Python 2.6: Celery series 3.1 or earlier. Python 2.5: Celery series 3.0 or earlier. Python 2.4 was Celery series 2.2 or earlier. Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform.
2.使用场景
异步任务:将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件,消息推送,音视频处理等等
定时任务:定时执行某件事情,比如每天数据统计
3.Celery的安装配置
pip install celery 消息中间件:RabbitMQ/Redis app=Celery('人物名',backend='xxx',broker='xxx')
4.Celery执行异步任务
基本使用
创建项目:celerytest
创建py文件:tasks.py
from celery import Celery import time broker = 'redis://127.0.0.1:6379/1' backend = 'redis://127.0.0.1:6379/2' app = Celery('test',broker=broker,backend=backend) @app.task def add(x,y): return x+y
创建py文件:add_task.py,添加任务
from tasks import add result = add.delay(4,5) print(result.id)
创建py文件,run.py,执行任务,或者使用命令执行:celery worker -A tasks -l info
注:windows下:celery worker -A tasks -l info -P eventlet
from tasks import add if __name__ == '__main__': add.worker_main() # cel.worker_main(argv=['--loglevel=info')
创建py文件:result.py ,查看任务执行结果
from celery.result import AsyncResult from tasks import add async = AsyncResult(id="e919d97d-2938-4d0f-9265-fd8237dc2aa3", app=cel) if async.successful(): result = async.get() print(result) # result.forget() # 将结果删除 elif async.failed(): print('执行失败') elif async.status == 'PENDING': print('任务等待中被执行') elif async.status == 'RETRY': print('任务异常后正在重试') elif async.status == 'STARTED': print('任务已经开始被执行') 执行add_task.py,添加任务,并获取任务ID 执行run.py ,或者执行命令:celery worker -A tasks -l info -P eventlet 执行result.py 检查任务状态并获取结果
多任务结构
multi_celery
├── celery_task# celery相关文件夹
│ ├── celery.py # celery连接和配置相关文件,必须叫这个名字
│ └── tasks1.py # 所有任务函数
│ └── tasks2.py # 所有任务函数
├── result.py # 检查结果
└── add_task.py # 触发任务
celery.py
from celery import Celery # broker:消息中间人用redis broker='redis://127.0.0.1:6379/1' # 结果存储在redis中 backend='redis://127.0.0.1:6379/2' # 第一个参数是别名,可以随便写 # include=[] app=Celery('test',broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2']) # 时区 app.conf.timezone = 'Asia/Shanghai' # 是否使用UTC app.conf.enable_utc = False
task1.py
from .celery import app @app.task def add(x,y): return x+y
taks2.py
from .celery import app @app.task def write_file(s): with open('a.txt','a',encoding='utf-8')as f: f.write(s) return '写成功'
result.py
from celery.result import AsyncResult # 导入celery对象 from celery_task.celery import app async = AsyncResult(id="ac2a7e52-ef66-4caa-bffd-81414d869f85", app=app) if async.successful(): # 任务执行的结果,也就是返回值 result = async.get() print(result) # result.forget() # 将结果删除 elif async.failed(): print('执行失败') elif async.status == 'PENDING': print('任务等待中被执行') elif async.status == 'RETRY': print('任务异常后正在重试') elif async.status == 'STARTED': print('任务已经开始被执行')
add_task.py
from celery_task import task1 from celery_task import task2 # 往队列中添加一个2+3的任务 result=task1.add.delay(2,3) print(result.id) # 往队列中添加一个写文件的任务 result=task2.write_file.delay('lqz') print(result.id)
添加任务(执行add_task.py),开启worker:celery worker -A celery_task -l info -P eventlet,检查任务执行结果(执行result.py)
5 Celery执行定时任务
设定时间让celery执行一个任务
add_task.py
from celery_app_task import add from datetime import datetime # 方式一 # v1 = datetime(2019, 2, 13, 18, 19, 56) # print(v1) # v2 = datetime.utcfromtimestamp(v1.timestamp()) # print(v2) # result = add.apply_async(args=[1, 3], eta=v2) # print(result.id) # 方式二 ctime = datetime.now() # 默认用utc时间 utc_ctime = datetime.utcfromtimestamp(ctime.timestamp()) from datetime import timedelta time_delay = timedelta(seconds=10) task_time = utc_ctime + time_delay # 使用apply_async并设定时间 result = add.apply_async(args=[4, 3], eta=task_time) print(result.id)
类似于contab的定时任务
多任务结构中celery.py修改如下
from datetime import timedelta from celery import Celery from celery.schedules import crontab cel = Celery('tasks', broker='redis://127.0.0.1:6379/1', backend='redis://127.0.0.1:6379/2', include=[ 'celery_task.tasks1', 'celery_task.tasks2', ]) cel.conf.timezone = 'Asia/Shanghai' cel.conf.enable_utc = False cel.conf.beat_schedule = { # 名字随意命名 'add-every-10-seconds': { # 执行tasks1下的test_celery函数 'task': 'celery_task.tasks1.test_celery', # 每隔2秒执行一次 # 'schedule': 1.0, # 'schedule': crontab(minute="*/1"), 'schedule': timedelta(seconds=2), # 传递参数 'args': ('test',) }, # 'add-every-12-seconds': { # 'task': 'celery_task.tasks1.test_celery', # 每年4月11号，8点42分执行 # 'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4), # 'schedule': crontab(minute=42, hour=8, day_of_month=11, month_of_year=4), # 'args': (16, 16) # }, }
启动一个beat：celery beat -A celery_task -l info
启动work执行：celery worker -A celery_task -l info -P eventlet
6.Django中使用Celery
在项目中创建celeryconfig.py
import djcelery djcelery.setup_loader() CELERY_IMPORTS = ( 'app01.tasks', ) # 有些情况可以防止死锁 CELERYD_FORCE_EXECV = True # 设置并发worker数量 CELERYD_CONCURRENCY = 4 # 允许重试 CELERY_ACKS_LATE = True # 每个worker最多执行100个任务被销毁，可以防止内存泄漏 CELERYD_MAX_TASKS_PER_CHILD = 100 # 超时时间 CELERYD_TASK_TIME_LIMIT = 12 * 30
在app01目录下面创建tasks.py
from celery import task import time @task def add(x,y): time.sleep(3) return x+y
视图函数views.py
from django.shortcuts import render,HttpResponse # Create your views here. from app01 import tasks def test(request): result=tasks.add.delay(2,4) print(result.id) return HttpResponse('ok')
settings.py
INSTALLED_APPS = [ ... 'djcelery', 'app01' ] ... from djagocele import celeryconfig BROKER_BACKEND='redis' BROKER_URL='redis://127.0.0.1:6379/1' CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2'
起worker:
python3 manage.py celery worker