目录
- 1. celery 快速使用
- 2. Celery异步任务框架
- 3. 按装使用步骤
- 4. celery包结构
- 5. celery异步任务，延迟任务，定时任务
- 6. django中使用celery
- 7. 秒杀逻辑
- 8. 双写一致性
- 8.2 celery定时任务实现双写一致性
1. celery 快速使用
1. Celery 官网：http://www.celeryproject.org/
2. Celery 官方文档英文版：http://docs.celeryproject.org/en/latest/index.html
3. Celery 官方文档中文版：http://docs.jinkan.org/docs/celery/
4. Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，专注于实时处理的异步任务队列，同时也支持任务调度
5. 注意：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异步任务框架
1. 可以不依赖任何服务器，通过自身命令，启动服务
2. celery服务为为其他项目服务提供异步解决任务需求的
注：会有两个服务同时运行，一个是项目服务，一个是celery服务，项目服务将需要异步处理的任务交给celery服务，celery就会在需要时异步完成项目的需求
例：人是一个独立运行的服务（django） | 医院也是一个独立运行的服务(celery)
正常情况下，人可以完成所有健康情况的动作，不需要医院的参与；但当人生病时，就会被医院接收，解决人生病问题
人生病的处理方案交给医院来解决，所有人不生病时，医院独立运行，人生病时，医院就来解决人生病的需求
3. 按装使用步骤
1. 安装：pip install celery
2. 使用步骤
1. 写一个main.py：实例化得到app对象，写函数，任务，注册成celery的任务
2. 再别的程序中：提交任务--->提交到broker中去了add.delay(3,4)
3. 启动worker，从broker中取任务执行，执行完放到backend中
4. win:
celery worker -A main -l info -P eventlet # 4.x及之前用这个
celery -A main worker -l info -P eventlet # 5.x及之后用这个
lin,mac:
celery worker -A main -l info
celery -A main worker -l info
5. 再backend中查看任务执行的结果
通过代码查看
# 查询执行完的结果
from main import app
from celery.result import AsyncResult
id = '89d05c2b-91ff-4c3e-b7c9-c9b3acd0be8f'
if __name__ == '__main__':
res = AsyncResult(id=id, app=app)
if res.successful():
result = res.get() #7
print(result)
elif res.failed():
print('任务失败')
elif res.status == 'PENDING':
print('任务等待中被执行')
elif res.status == 'RETRY':
print('任务异常后正在重试')
elif res.status == 'STARTED':
print('任务已经开始被执行')
main.py
from celery import Celery
import time
backend = 'redis://127.0.0.1:6379/1'
broker = 'redis://127.0.0.1:6379/0'
app = Celery('test', backend=backend, broker=broker)
# 写任务，任务就是函数，加个装饰器，变成celery的任务
@app.task
def add(a, b):
time.sleep(2) # 假设任务耗时比较久
print(a + b)
return a + b
s1.py
from main import add
print('hello world')
# 执行同步任务
# res = add(3, 4)
# print(res)
# 执行异步任务，把任务提交到celery中了
res = add.delay(4, 4)
print(res) # 89d05c2b-91ff-4c3e-b7c9-c9b3acd0be8f 唯一的uuid
get.result.py
# 查询执行完的结果
from main import app
from celery.result import AsyncResult
id = '89d05c2b-91ff-4c3e-b7c9-c9b3acd0be8f'
if __name__ == '__main__':
res = AsyncResult(id=id, app=app)
if res.successful():
result = res.get() #7
print(result)
elif res.failed():
print('任务失败')
elif res.status == 'PENDING':
print('任务等待中被执行')
elif res.status == 'RETRY':
print('任务异常后正在重试')
elif res.status == 'STARTED':
print('任务已经开始被执行')
s2.py
def wrapper(func):
def inner(*args, **kwargs):
res = func(*args, **kwargs)
return res
inner.delay = 'xxx'
return inner
@wrapper
def add():
print('add')
print(add.delay)
4. celery包结构
1. 写一个celery的包，以后，再任意项目中，想用，把包copy进去，导入使用即可
celery_task
__init__.py
celery.py
user_task.py
home_task.py
add_task.py
get_resullt.py
2. 使用步骤
新建包：celery_task
在包先新建一个 celery.py
在里面写app的初始化
在包里新建user_task.py 编写用户相关任务
在包里新建home_task.py 编写首页相关任务
其它程序，提交任务
启动worker --->它可以先启动，在提交任务之前-->包所在的目录下
celery -A celery_task worker -l info -P eventlet
查看任务执行的结果了
celery_task/celery.py
from celery import Celery
backend = 'redis://127.0.0.1:6379/1'
broker = 'redis://127.0.0.1:6379/0'
# 一定不要忘了include
app = Celery(__name__, broker=broker, backend=backend, include=['celery_task.home_task', 'celery_task.user_task'])
celery_task/home_task.py
import time
from .celery import app
@app.task
def add(a, b):
time.sleep(3)
print('计算结果是：%s' % (a + b))
return a + b
celery_task/user_task.py
import time
from .celery import app
@app.task
def send_sms(mobile, code):
time.sleep(1)
print('短信发送成功：%s,验证吗是%s' % (mobile, code))
return True
add_task.py
from celery_task.user_task import send_sms
from celery_task.home_task import add
# 提交了一个发送短信异步任务
res = send_sms.delay('21341414131', '9999')
print(res) # 89d05c2b-91ff-4c3e-b7c9-c9b3acd0be8f
get_result.py
# 查询执行完的结果
from celery_task.celery import app
from celery.result import AsyncResult
id = '89d05c2b-91ff-4c3e-b7c9-c9b3acd0be8f'
if __name__ == '__main__':
res = AsyncResult(id=id, app=app)
if res.successful():
result = res.get() #7
print(result)
elif res.failed():
print('任务失败')
elif res.status == 'PENDING':
print('任务等待中被执行')
elif res.status == 'RETRY':
print('任务异常后正在重试')
elif res.status == 'STARTED':
print('任务已经开始被执行')
5. celery异步任务，延迟任务，定时任务
5.1 异步任务
任务.delay(参数，参数)
# 提交一个add的异步任务
res=add.delay(5,6)
print(res)
5.2 延迟任务
任务.apply_async(args=[参数，参数],eta=时间对象(utc时间))
# 提交一个延迟任务 10s后执行
# eta 时间对象 10s后的时间对象
from datetime import datetime, timedelta
#
# 得到5s后的utc时间,celery默认使用utc时间
eta = datetime.utcnow() + timedelta(seconds=10)
# local_eta = datetime.now() + timedelta(seconds=5)
# print(eta)
# print(local_eta)
res=add.apply_async(args=(200, 50), eta=eta)
print(res)
5.3 定时任务
# 写定时任务
# app的配置项
# 时区 app.conf 所有的配置字典，默认的
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
## 编写定时任务
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'send_sms_task': {
'task': 'celery_task.user_task.send_sms',
'schedule': timedelta(seconds=5),
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
'args': ('1897334444', '7777'),
},
'add_task': {
'task': 'celery_task.home_task.add',
'schedule': crontab(hour=20, minute=20, day_of_week=3), # 每周一早八点
'args': (10, 20),
}
}
1 app的配置文件中配置
2 启动worker :干活的人
celery -A celery_task worker -l info -P eventlet
3 启动beat ：提交任务的人
celery -A celery_task beat -l info
6. django中使用celery
使用步骤:
1 把写好的包复制到项目路径下
2 在包内的celery.py 的上面加入代码
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffy_api.settings.dev')
import django
django.setup()
3 在django的视图类中，导入，提交任务
4 启动worker，beat
views.py
from celery_task.user_task import send_sms
def index(request):
mobile = request.GET.get('mobile')
res = send_sms.delay(mobile, '8888')
print(res)
return HttpResponse('已经发送了')
celery_task/user_task
import time
from .celery import app
import random
@app.task
def send_sms(mobile, code):
from libs.send_tx_sms import send_sms_by_phone
from user.models import UserInfo
send_sms_by_phone(mobile, code)
user = UserInfo.objects.all().filter(mobile=mobile).first()
print('给%s发送短信，短信发送成功：%s，验证吗是%s' % (user.username, mobile, code))
return True
celery_task/celery.py
from celery import Celery
# django 中集成celery，需要加入
import os
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffy_api.settings.dev')
import django
django.setup()
backend = 'redis://127.0.0.1:6379/1'
broker = 'redis://127.0.0.1:6379/0'
# 一定不要忘了include
app = Celery(__name__, broker=broker, backend=backend, include=['celery_task.home_task', 'celery_task.user_task'])
# 写定时任务
# app的配置项
# 时区 app.conf 所有的配置字典，默认的
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
## 编写定时任务
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
# 'send_sms_task': {
# 'task': 'celery_task.user_task.send_sms',
# 'schedule': timedelta(hours=5),
# # 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八点
# 'args': ('1897334444', '7777'),
# },
'update_banner': {
'task': 'celery_task.home_task.update_banner',
'schedule': timedelta(seconds=50),
'args': (),
}
}
7. 秒杀逻辑
前端
1 秒杀按钮
2 事件：向后端秒杀接口发送请求，发送完立马起了一个定时任务，每个5s，向后端查看一下是否秒杀成功，如果秒杀没成功，定时任务继续执行，如果秒杀成功了，清空定时任务，弹窗告诉他
<template>
<div>
<img src="" alt="" height="300px" width="300px">
<br>
<el-button type="danger" plain @click.once="handleClick">秒杀</el-button>
</div>
</template>
<script>
export default {
name: "Seckill",
methods: {
handleClick() {
this.$axios.get(this.$settings.BASE_URL + 'userinfo/seckill/').then(res => {
if (res.data.code == 100) {
let task_id = res.data.id
this.$message({
message: res.data.msg,
type: 'error'
});
// 起个定时任务，每隔5s向后端查询一下是否秒杀成功
let t = setInterval(() => {
this.$axios.get(this.$settings.BASE_URL + 'userinfo/get_result/?id=' + task_id).then(
res => {
if (res.data.code == 100 || res.data.code == 101) { //秒杀结束了，要么成功，要么失败了
alert(res.data.msg)
// 销毁掉定时任务
clearInterval(t)
} else if (res.data.code == 102) {
//什么事都不干
}
}
)
}, 5000)
}
})
}
}
}
</script>
<style scoped>
</style>
后端：
秒杀接口
提交秒杀任务
views.py
from celery_task.user_task import seckill_task
from celery_task.celery import app
from celery.result import AsyncResult
def seckill(request):
# 提交秒杀任务
res = seckill_task.delay()
return JsonResponse({'code': 100, 'msg': '正在排队', 'id': str(res)})
def get_result(request):
task_id = request.GET.get('id')
res = AsyncResult(id=task_id, app=app)
if res.successful():
result = res.get() # 7
return JsonResponse({'code': 100, 'msg': str(result)})
elif res.failed():
print('任务失败')
return JsonResponse({'code': 101, 'msg': '秒杀失败'})
elif res.status == 'PENDING':
print('任务等待中被执行')
return JsonResponse({'code': 102, 'msg': '还在排队'})
user_task.py
@app.task
def seckill_task():
time.sleep(6) # 模拟秒杀需要3s
res = random.choice([True, False])
if res:
return '秒杀成功'
else:
return '很遗憾，您没有秒到'
8. 双写一致性
8.1 接口加缓存
首页轮播图接口，加缓存
提交了接口的响应速度
提高并发量
home/views.py
from django.core.cache import cache
from django.conf import settings
from utils.response import APIResponse
class BannerView(GenericViewSet, CommonListModelMixin):
queryset = Banner.objects.all().filter(is_delete=False, is_show=True).order_by('orders')[:settings.BANNER_COUNT]
serializer_class = BannerSerializers
def list(self, request, *args, **kwargs):
result = cache.get('banner_list')
if result: # 缓存里有
print('走了缓存，速度很快')
return APIResponse(result=result)
else:
# 去数据库拿
print('走了数据库，速度慢')
res = super().list(request, *args, **kwargs)
result = res.data.get('result') # {code:100,msg:成功，result:[{},{}]}
cache.set('banner_list', result)
return res
加了缓存，如果mysql数据变了，由于请求的都是缓存的数据，导致mysql和redis的数据不一致
双写一致性问题
1 修改mysql数据库，删除缓存 【缓存的修改是在后】
2 修改数据库，修改缓存 【缓存的修改是在后】
3 定时更新缓存 ---》针对于实时性不是很高的接口适合定时更新
给首页轮播图接口加入了缓存，出现了双写一致性问题，使用定时更新来解决双写一致性的问题【会存在不一致的情况，我们可以忽略】---》定时任务，celery的定时任务
8.2 celery定时任务实现双写一致性
home_task.py
@app.task
def update_banner():
# 更新缓存
# 查询出现在轮播图的数据
queryset = Banner.objects.all().filter(is_delete=False, is_show=True).order_by('orders')[:settings.BANNER_COUNT]
ser = BannerSerializer(instance=queryset, many=True)
# ser 中得图片，没有前面地址
for item in ser.data:
item['image'] = settings.HOST_URL + item['image']
cache.set('banner_list', ser.data)
return True
celery.py
app.conf.beat_schedule = {
'update_banner': {
'task': 'celery_task.home_task.update_banner',
'schedule': timedelta(seconds=50),
'args': (),
}
}
dev.py
HOST_URL='http://127.0.0.1:8000'
启动django
启动worker
celery -A celery_task worker -l info -P eventlet
启动beat
celery -A celery_task beat -l info
第一次访问：查的数据库，放入了缓存
以后再访问，走缓存
一旦mysql数据改了，缓存可能不一致
当时我们定时更新，最终保持了一致
