一、 celery介绍
Celery 官网:http://www.celeryproject.org/ Celery 官方文档英文版:http://docs.celeryproject.org/en/latest/index.html
?celery能干什么:异步任务,延迟任务,定时任务
-异步执行:解决耗时任务,将耗时操作任务提交给Celery去异步执行,比如发送短信/邮件、消息推送、音视频处理等等、
-延迟执行:解决延迟任务
-定时执行:解决周期(周期)任务,比如每天数据统计
"""
1)可以不依赖任何服务器,通过自身命令,启动服务
2)celery服务为为其他项目服务提供异步解决任务需求的
注:会有两个服务同时运行,一个是项目服务,一个是celery服务,项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求
人是一个独立运行的服务 | 医院也是一个独立运行的服务
正常情况下,人可以完成所有健康情况的动作,不需要医院的参与;但当人生病时,就会被医院接收,解决人生病问题
人生病的处理方案交给医院来解决,所有人不生病时,医院独立运行,人生病时,医院就来解决人生病的需求
"""
Celery is a project with minimal funding, so we don’t support Microsoft Windows. Please don’t open any issues related to that platform
Sanic:python3.5以上,不支持win
-选择使用mac开发
-装乌班图
-win远程连接linux开发
uwgi(不支持win)+django
二、 celery架构
Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括, Redis等等
Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中(本质,一个work就是一个进程)。
Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括 redis等
三、 celery基本使用
pip install celery
pip install eventlet
celery worker -A celery_task -l info -P eventlet
pip install eventlet
celery -A celery_task worker -l info -P eventlet
celery worker -A celery_task -l info
celery -A celery_task worker -l info
celery beat -A celery_task -l info
两种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
注:包名随意
3.1 普通使用
新建celery_task.py
from celery import Celery
backend='redis://127.0.0.1:6379/1'
broker='redis://127.0.0.1:6379/2'
app=Celery('test',broker=broker,backend=backend)
@app.task
def add(a,b):
import time
time.sleep(3)
return a+b
新建add_task.py添加任务
from celery_task import add
res=add.delay(3,4)
print(res)
新建get_result.py查询任务
from celery_task import app
from celery.result import AsyncResult
id = 'b20f827d-dcf5-4fdf-a646-5abb963dc1d3'
if __name__ == '__main__':
a = AsyncResult(id=id, app=app)
if a.successful():
result = a.get()
print(result)
elif a.failed():
print('任务失败')
elif a.status == 'PENDING':
print('任务等待中被执行')
elif a.status == 'RETRY':
print('任务异常后正在重试')
elif a.status == 'STARTED':
print('任务已经开始被执行')
启动worker
pip install eventlet
celery -A celery_task worker -l info -P eventlet
celery -A celery_task worker -l info
3.2 包结构
celery_task
├── __init__.py
├── user_task.py
├── order_task.py
├── home_task.py
└── celery.py
celery.py
from celery import Celery
backend = 'redis://127.0.0.1:6379/1'
broker = 'redis://127.0.0.1:6379/2'
app = Celery('test', broker=broker, backend=backend, include=[
'celery_task.home_task',
'celery_task.order_task',
'celery_task.user_task'
])
print(app.conf)
app.conf.timezone = 'Asia/Shanghai'
app.conf.enable_utc = False
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'send_sms_5': {
'task': 'celery_task.user_task.send_sms',
'schedule': timedelta(seconds=5),
'args': (189533333,),
},
'add_3': {
'task': 'celery_task.home_task.add',
'schedule': timedelta(seconds=3),
'args': (6,8),
}
}
home_task.py
from .celery import app
@app.task
def add(a,b):
return a+b
order_task.py
from .celery import app
@app.task
def write_file(s):
with open(r'D:\lqz2\luffy_api\log\a.txt','w',encoding='utf-8') as f:
f.write(s)
return True
user_task.py
from .celery import app
@app.task
def send_sms(s):
import time
time.sleep(3)
return '%s手机号,短信发送成功'%s
四、 celery执行异步任务,延迟任务,定时任务
3.1 异步任务
from celery_task.user_task import send_sms
res=send_sms.delay('18953675221')
print(res)
3.2 延迟任务
from celery_task.user_task import send_sms
from datetime import datetime, timedelta
eta=datetime.utcnow() + timedelta(seconds=10)
res=send_sms.apply_async(args=['1888888',],eta=eta)
print(res)
3.3定时任务
app.conf.timezone = 'Asia/Shanghai'
app.conf.enable_utc = False
?
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'task-mul': {
'task': 'celery_task.user_task.mul',
'schedule': timedelta(seconds=3),
'args': (3, 15),
},
'task-add': {
'task': 'celery_task.home_task.add',
'schedule': timedelta(seconds=10),
'args': (3, 5),
},
}
celery beat -A celery_task -l info
celery -A celery_task worker -l info -P eventlet
五、 django中使用celery
-django-celery
-django
-celery
六、 首页轮播图接口加入缓存
1 如果mysql数据变化了,轮播图表数据变化了,由于缓存没有更新,就会出现问题
2 双写一致性问题
-因为使用了缓存,mysql和redis的数据不一致了
-如果解决:(缓存更新策略)
-加入过期时间? 60s? 过期时间内数据还有问题
-定时更新缓存(每隔60s,更新一下缓存)
-数据库一更改,立马改缓存
-先改数据库,再改缓存
-先改缓存,再改数据库
-数据库一改,删除缓存
-先删缓存,再改数据库
-先改数据库,再删缓存(稍微好一些)
3 首页轮播图接口:缓存更新策略是定时更新
-celery的定时任务
-使用celery的定时任务,完成首页轮播图缓存的更新
|