目录一、celery介绍二、celery架构消息中间件任务执行单元任务结果存储使用场景三、celery使用方式一:在一个文件夹内的三个页面方式二:worker单独做一个项目文件,添加任务和获取结果分离出来(执行异步任务)执行延迟任务添加定时任务四、django中配置celery
一、celery介绍
Celery 官网:http://www.celeryproject.org/
Celery 官方文档英文版:http://docs.celeryproject.org/en/latest/index.html
Celery 官方文档中文版:http://docs.jinkan.org/docs/celery/
异步任务框架
执行异步任务
执行延迟任务
执行定时任务
二、celery架构
# 如果 Celery对象:Celery(...) 是放在一个模块下的
# 1)终端切换到该模块所在文件夹位置:scripts
# 2)执行启动worker的命令:celery worker -A 模块名 -l info -P eventlet
# 注:windows系统需要eventlet支持,Linux与MacOS直接执行:celery worker -A 模块名 -l info
# 注:模块名随意
# 如果 Celery对象:Celery(...) 是放在一个包下的
# 1)必须在这个包下建一个celery.py的文件,将Celery(...)产生对象的语句放在该文件中
# 2)执行启动worker的命令:celery worker -A 包名 -l info -P eventlet
# 注:windows系统需要eventlet支持,Linux与MacOS直接执行:celery worker -A 模块名 -l info
# 注:包名随意
消息中间件
Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括,RabbitMQ, Redis等等
任务执行单元
Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中。
任务结果存储
Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, redis等
使用场景
异步执行:解决耗时任务,将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等等
延迟执行:解决延迟任务
定时执行:解决周期(周期)任务,比如每天数据统计
三、celery使用
安装:pip install celery
根据celery架构,我们可以看出,worker就像是一个工人,一直在工作,如果有工作的时候就会从broker(Redist)里去拿过来执行,并放到pakend(redist)中存放结果
方式一:在一个文件夹内的三个页面
worker执行页面
import celery
# broker存储的位置
broker = 'redis://127.0.0.1:6379/1'
# backend存储的位置
backend ='redis://127.0.0.1:6379/2'
# 实例化的celery对象
app=celery.Celery(__name__,broker=broker,backend=backend)
# 需要添加的任务
@app.task
def add(x,y):
print(x*y)
return x+y
broker提交任务的页面
from celery_test import add
# 执行这个文件,就是把这个任务添加到数据库中,只要worker在工作
# 就会把这个任务从数据库1中拿出来执行,并把结果放到数据库2中
ret = add.delay(3,4)
# ret 是这个任务的uuid,用于获取任务结果
backend获取任务结果的页面
from celery_test import app
from celery.result import AsyncResult
# 任务对象的唯一标识:uuid
id = '19dc2faa-39f9-47b6-af77-e9d3a4d05d2e'
if __name__ == '__main__':
async1 = AsyncResult(id=id, app=app)
if async1.successful():
result = async1.get()
print(result)
elif async1.failed():
print('任务失败')
elif async1.status == 'PENDING':
print('任务等待中被执行')
elif async1.status == 'RETRY':
print('任务异常后正在重试')
elif async1.status == 'STARTED':
print('任务已经开始被执行')
方式二:worker单独做一个项目文件,添加任务和获取结果分离出来(执行异步任务)
创建一个celery项目(包),内部必须含有名字为celery的py文件,在内部创建celery对象
celery.py
import celery
broker = 'redis://127.0.0.1:6379/1'
backend ='redis://127.0.0.1:6379/2'
app=celery.Celery(__name__,broker=broker,backend=backend,include=['celery_task.task1','celery_task.task2'])
编写需要添加的任务也在这个包内,可创建不同的任务文件,可添加多个
task1.py
from .celery import app
@app.task
def add(x,y):
print(x,y)
return x+y
执行延迟任务
添加任务页面
# 执行延迟任务就是多个一个时间参数
# 这里注意,时间参数是根据utc时间,并不是中国时间
from datetime import datetime, timedelta
# 时间对象必须和时间对象相加
eta=datetime.utcnow() + timedelta(seconds=10)
add.apply_async(args=(200, 50), eta=eta)
添加定时任务
celery页面
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'low-task': {
'task': 'celery_task.tasks.low',
'schedule': timedelta(seconds=3),
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (300, 150),
}
}
# 定时任务的添加必须要新启动一个beat命令去工作
# celery beat -A celery_task -l info
四、django中配置celery
celery包最好放在根路径下
添加定时任务
celery.py
import os
# 配置django配置
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
# 配置celery的worker环境
from celery import Celery
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# 实例化worker对象app,用include添加定时任务
app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.task2'])
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'low-task': {
'task': 'celery_task.task2.update_banner',
'schedule': timedelta(seconds=10),
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
# 'args': (),
}
}
# 定时任务的添加必须要新启动一个beat命令去工作
# celery beat -A celery_task -l info
task定时任务
from .celery import app
from django.core.cache import cache
# 每三秒跟新一次缓存
# 添加一个任务,celery内设定间隔时间3秒
# 任务内去cache跟新banner_list,从数据库中拿到banner放到cache中的banner_list中
from home.models import Banner
# from luffyapi.settings import const
# [OrderedDict(
# [('image', 'http://127.0.0.1:8000/media/banner/banner1.png'), ('link', '/free-courses'), ('name', 'banner1')]),
# OrderedDict(
# [('image', 'http://127.0.0.1:8000/media/banner/banner2.png'), ('link', '/light-courses'), ('name', 'banner2')]),
# OrderedDict(
# [('image', 'http://127.0.0.1:8000/media/banner/banner3.png'), ('link', '/actual-courses'), ('name', 'banner3')])]
from home.sers import BannerSer
@app.task
def update_banner(*args, **kwargs):
queryset = Banner.objects.filter(is_delete=False)
banner_ser = BannerSer(queryset,many=True)
for banner in banner_ser.data:
banner['image'] = 'http://127.0.0.1:8000' + banner['image']
cache.set('banner_list', banner_ser.data)
print(banner_ser.data)
return '更新banner成功'
最新评论