Celery在Django中的简单应用
- 1、celery 架构
Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列。同时也支持任务调度。需要注意的是,celery并不支持Windows,所以Windows相关的问题很可能在官方无法得到回应。
Celery的架构由三部分组成,消息中间件(message broker)、任务执行单元(worker)和 任务执行结果存储(task result store)组成。
1)可以不依赖任何服务器,通过自身命令,启动服务(内部支持socket) 2)celery服务为为其他项目服务提供异步解决任务需求的 注:会有两个服务同时运行,一个是项目服务,一个是celery服务,项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求。
使用场景: 异步执行:解决耗时任务,将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等等。 延迟执行:解决延迟任务。 定时执行:解决周期(周期)任务,比如每天数据统计。
2、celery 的简单使用
首先需要安装celery模块:pip install celery
2.1 最简单的使用
- 写一个py文件(celery_task.py):
from celery import Celery
broker='redis://127.0.0.1:6379/1' #broker任务队列
backend='redis://127.0.0.1:6379/2' # 结构存储,执行完的结果存在这
app=Celery(__name__,broker=broker,backend=backend)
#添加任务逻辑
@app.task
def add(x,y):
print(x,y)
return x+y
- 启动worker
# cmd或者Terminal中用命令来执行
# 非windows
celery worker -A celery_task -l info
# windows:
pip install eventlet
celery worker -A celery_task -l info -P eventlet
- 发送任务
from celery_task import add
ret=add.delay(5,4) #向broker中添加一个任务
print(ret) # ret是celery返回的任务id号,可以使用该ID号取回任务处理的结果
- 查看任务执行结果
from celery_task import app
from celery.result import AsyncResult
id = '3e397fd7-e0c1-4c5c-999c-2655a96793bb' # 此处的ID即为发送任务时celery返回的任务id。
if __name__ == '__main__':
async = AsyncResult(id=id, app=app)
'''
在Python3.7及以后的版本中async成为了保留关键字,建议使用3.7以下版本或者更换该变量名。
'''
if async.successful():
result = async.get()
print(result)
elif async.failed():
print('任务失败')
elif async.status == 'PENDING':
print('任务等待中被执行')
elif async.status == 'RETRY':
print('任务异常后正在重试')
elif async.status == 'STARTED':
print('任务已经开始被执行')
2.2 包结构的使用
- 新建一个包(celery_task)
# 包结构
-celery_task
-__init__.py
-celery.py # 该py文件一点要叫celery.py
-tasks.py
- celery.py
from celery import Celery
broker='redis://127.0.0.1:6379/1'
backend='redis://127.0.0.1:6379/2'
app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])
- tasks.py
from .celery import app
@app.task
def add(x,y):
print(x,y)
return x+y
@app.task
def mutile(x,y):
print(x,y)
return x*y
- 启动worker
- 发送任务
from celery_task.tasks import add,mutile
# 提交异步任务
ret=add.delay(6,7)
print(ret)
# 提交延迟任务
from datetime import datetime, timedelta
# 需要utc时间
eta=datetime.utcnow() + timedelta(seconds=10)
ret=mutile.apply_async(args=(240, 50), eta=eta) # 延迟10s执行mutile任务,args为其传值。
print(ret)
- 根据返回的id取处理结果
2.3 celery执行定时任务
在celery_task 包的celery.py 中:
from celery import Celery
from datetime import timedelta
from celery.schedules import crontab
broker='redis://127.0.0.1:6379/1'
backend='redis://127.0.0.1:6379/2'
app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])
# 执行定时任务
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False # 不使用utc时间
# 任务的定时配置
app.conf.beat_schedule = {
'add-task': {
'task': 'celery_task.tasks.add',
# 'schedule': timedelta(seconds=3), # 每隔3秒
'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': (300, 150), # 传参
}
}
除了需要启动worker,由于是定时自动提交,还需要启动beat。
-celery worker -A celery_task -l info -P eventlet
-celery beat -A celery_task -l info
2.4 celery 在Django中的使用
在Django项目的根目录新建一个celery_task包。其中celery.py示例:
from celery import Celery
# 加载django环境,否则无法使用Django的model或其他内容
import os
import django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev") # "luffyapi.settings.dev"为项目配置文件,需要注意路径
django.setup()
broker='redis://127.0.0.1:6379/1'
backend='redis://127.0.0.1:6379/2'
app=Celery(__name__,broker=broker,backend=backend,include=['celery_task.tasks',])
# 执行定时任务
# 时区
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任务的定时配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'add-task': {
'task': 'celery_task.tasks.update',
'schedule': timedelta(seconds=30), # 定时30秒执行刷新任务,将数据库中的数据缓存到Redis中
}
}
tasks.py:
from .celery import app
@app.task
def update():
from home import serializer
from home import models
from django.conf import settings
from django.core.cache import cache
queryset_banner = models.Banner.objects.filter(is_delete=False,is_show=True).order_by('display_order')[
:settings.BANNER_COUNTER]
serializer_banner=serializer.BannerModelSerilaizer(instance=queryset_banner,many=True)
cache.set('banner_list',serializer_banner.data)
return True
其中celery在Django项目中的使用最重要的就是加载Django的环境,以便于celery在处理任务的时候可以使用Django的model及其他内容。其中tasks只要遵循官网的目录结果,可以在Django的每个APP下建立一个tasks.py,celery可以自动识别。
"""
celery框架django项目工作流程
1)加载django配置环境
2)创建Celery框架对象app,配置broker和backend,得到的app就是worker
3)给worker对应的app添加可处理的任务函数,用include配置给worker的app
4)完成提供的任务的定时配置app.conf.beat_schedule
5)启动celery服务,运行worker,执行任务
6)启动beat服务,运行beat,添加任务
重点:由于采用了django的反射机制,使用celery.py所在的celery_task包必须放置项目的根目录下
"""
Author:Laoqi
- python爬取链家租房之获取房屋的链接和页面的详细信息
- 信用卡“坏账”客户分析(一)
- 一道简单的sql语句题
- python爬虫反爬取---设置User Agent自动变换header文件
- 一文读懂Python多线程
- 深入理解Python变量作用域与函数闭包
- TensorFlow从1到2 - 5 - 非专家莫入!TensorFlow实现CNN
- JetBrains Rider 破解 (ideaIU等等开发工具都通用)
- python中的小魔法(一)
- 由问题入手,步步爬出Python中赋值与拷贝的坑
- python爬取链家租房之获得每一页的房屋信息地址(持续更新)
- python使用正则表达式
- python在租房过程中的应用
- python爬虫反爬取---设置IP代理自动变换requests.get()中proxy的IP
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- GLMM:广义线性混合模型(遗传参数评估)
- 特征锦囊:今天一起搞懂机器学习里的L1与L2正则化
- 【一天一大 lee】二叉搜索树的最近公共祖先 (难度:简单) - Day2020092
- Spring多数据源事务如何玩? | Spring系列46篇
- 使用Mfuzz包做时间序列分析
- 网络安全 | 瑞哥带你全方位解读防火墙技术!
- 【SpringBoot DB 系列】Jooq 之新增记录使用姿势
- 突击并发编程JUC系列-并发工具 Semaphore
- 构建Linux根文件系统
- ARM指令ldr和adr的区别
- 挂载文件系统出现"kernel panic..." 史上最全解决方案
- 汇编程序调用c函数为什么需要设置栈?
- Uboot到底如何启动内核
- Flink部署及作业提交(On YARN)
- Flink部署及作业提交(On Flink Standalone)