Celery
1、什么是Celery
- Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，专注于实时处理的异步任务队列，同时也支持任务调度。
- 用Python写的执行 定时任务和异步任务的框架
执行异步任务：
- 创建任务：tasks.py
- 把任务添加到队列中：add_task.py
- 开启work，执行任务
- 用命令：celery -A tasks worker -l info
- 在 Windows下：celery -A tasks worker -l info -P eventlet
- 查看任务结果：task_resut.py
多任务结构：
- 重点：执行work的时候：celery -A tasks worker -l info -P eventlet
2、Celery架构
Celery的架构由三部分组成，消息中间件，任务执行单元和任务执行结果存储(task result store )组成。
消息中间件
Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成。包括，RabbitMQ, Redis等等
任务执行单元
Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中。
任务结果存储
Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括AMQP, redis等
3、使用场景
异步任务：将耗时操作任务提交给Celery取异步执行 ，比如发送 短信、邮件、消息推送、音频/视频处理等等
定时任务：定时执行某件事情，比如每天数据统计
4、Celery的安装配置
pip install celery
app=Celery('任务名', backend='xxx',broker='xxx')
5、Celery执行异步任务
基本使用
创建项目：celerytest
创建py文件：task.py
`from celery import Celery
import time
# broker:消息中间人用redis
broker = 'redis://127.0.0.1:6397/1'
# 结果存储在redis中
backend = 'redis://127.0.0.1:6379/2'
# 第一个参数是别名，可以随便写
app = Celery('test', broker=broker, backend=backend)
@app.task
def add(x, y):
time.sleep(3)
return x + y
创建py文件：add_task.py,添加任务
import task
if __name__ == '__main__':
# 之前这样写，直接就执行 函数
task.add()
# 现在把函数添加到执行队列中，参数写在delay中
# result不是函数的执行结果，他是个对象
result = task.add.delay(2, 3)
# 这个任务唯一的id
print(result.id)
创建py文件：result.py，查看任务执行结果
from celery.result import AsyncResult
from task 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,添加任务，并获取任务ID
执行命令：celery worker -A celery_app_task -l info -P eventlet
执行result.py检查任务状态并获取结果
6、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)
启动一个beat: celery beat -A celery_task -l info
启动work执行：celery worker -A celery_task -l info -P eventlet
7、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
@task
def add(a,b):
with open('a.text', 'a',encoding='utf-8') as f:
f.write('a')
print(a+b)
视图函数views.py
from django.shortcuts import render,HttpResponse
from app01.tasks import add
from datetime import datetime
def test(request):
# result=add.delay(2,3)
ctime = datetime.now()
# 默认用utc时间
utc_ctime = datetime.utcfromtimestamp(ctime.timestamp())
from datetime import timedelta
time_delay = timedelta(seconds=5)
task_time = utc_ctime + time_delay
result = add.apply_async(args=[4, 3], eta=task_time)
print(result.id)
return HttpResponse('ok')
settings.py
INSTALLED_APPS = [
...
'djcelery',
'app01'
]
...
from django_celery import celeryconfig
BROKER_BACKEND='redis'
BOOKER_URL='redis://127.0.0.1:6379/1'
CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/2'