作者：大家好，我是阿旭
支持我：点赞收藏⭐️留言
GitHub主页：Tengxu666 欢迎star
- celery是什么
- celery的应用场景
- celery添加到Django项目
- 发布异步和延时任务
- 发布定时任务
- 实现秒杀系统的消息队列
Celery 是一个 基于python开发的分布式异步消息任务队列，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到异步任务，就可以考虑使用celery， 举几个实例场景中可用的例子:
- 异步任务：将耗时的操作任务提交给Celery去异步执行。
- 做一个定时、延时任务。
- 处理消息队列。
Celery 在执行任务时需要通过一个消息中间件（Broker）来接收和发送任务消息，以及存储任务结果， 一般使用rabbitMQ or Redis，我们这里使用的是redis
celery的应用场景- 比如你需要发送验证码时，为了能有更快响应你可以使用celery实现异步发送，两个操作同时进行。
- 如果你的项目中需要定时任务时，比如说定时爬取一些内容、定时清理一些数据，你可以使用celery来设置定时任务
- 如果要处理高并发，比如说秒杀活动，你可以将用户的全部请求放在消息队列中，依次执行，每个任务之间不会相互影响。
接下来我们看一下具体的实现。
celery添加到Django项目大家来看一下我的项目结构
具体项目代码可访问GitHub：Tengxu666 欢迎star
drf_celery
--- celery_tasks
--- _async
--- tasks.py
--- config.py
--- main.py
--- drf_celery
--- asgi.py
--- setting.py
--- urls.py
--- wsgi.py
--- message
--- apps.py
--- models.py
--- views.py
其中最主要的就是celery_tasks这个文件夹，它是用来配置celery的。其中async/tasks.py是用来放置一些具体的celery任务。config.py是做了一些celery的配置，比如说消息传输中间件、消息队列存储以及定时任务的配置。main.py是引用了config文件中的配置实例化celery对象。
config.py
# celery_tasks/config.py
# 配置一个 config.py, 存储配置信息, 实现 配置信息 存在于 单独的配置文件中
# 之后在main.py, 让实例对象 app 加载其中的配置
# 配置 broker 存储在 redis:14号 库
"""
broker: 是一个消息传输的中间件，可以理解为一个邮箱。每当应用程序调用celery的异步任务的时候，
会向broker传递消息，而后celery的worker将会取到消息，进行对于的程序执行。好吧，这个邮箱可以看成是一个消息队列。
"""
broker_url = 'redis://127.0.0.1/14'
# 配置 backend 存储在 redis:15号 库
"""
backend: 通常程序发送的消息，发完就完了，可能都不知道对方时候接受了。为此，celery实现了一个backend，
用于存储这些消息以及celery执行的一些消息和结果。对于 brokers，官方推荐是rabbitmq和redis，
至于backend，就是数据库啦。为了简单起见，我们都用redis。
"""
result_backend = 'redis://127.0.0.1/15'
# --------- 定时任务设置 -----------
# 指定任务序列化方式
task_serializer = 'json'
# 指定结果序列化方式
result_serializer = 'json'
# 指定任务接受的序列化类型.
accept_content = ['json']
timezone = "Asia/Shanghai" # 时区设置
worker_hijack_root_logger = False # celery默认开启自己的日志，可关闭自定义日志，不关闭自定义日志输出为空
result_expires = 60 * 60 * 24 # 存储结果过期时间（默认1天）
# 导入任务所在文件
imports = [
"celery_tasks._async.tasks"
]
# 需要执行任务的配置
beat_schedule = {
# 暂无定时任务
}
# "schedule": crontab（）与crontab的语法基本一致
# "schedule": crontab(minute="*/10", # 每十分钟执行
# "schedule": crontab(minute="*/1"), # 每分钟执行
# "schedule": crontab(minute=0, hour="*/1"), # 每小时执行
mian.py
# 需先导入工程配置文件
# 作用: 比如 获取 Django项目的 redis 配置?!
import os
if not os.getenv('DJANGO_SETTINGS_MODULE'):
# 设置 添加 Django项目 的 setting 路径 到 os 环境变量
os.environ['DJANGO_SETTINGS_MODULE'] = 'drf_celery.settings' # 要对应 自己的 Django 项目名
# 创建celery实例
# main 其实 就是 给celery设置一个名字, 这个名字唯一就可以
# 推荐使用 文件路径
from celery import Celery
app = Celery(main='celery_tasks')
# 加载 config.py 配置文件(设置broker任务队列)
# 配置见下文
app.config_from_object('celery_tasks.config')
# 实现 celery实例对象 自动检测任务
# 参数: 列表需要填写任务的包路径
app.autodiscover_tasks(['celery_tasks._async'])
"""
启动周期任务（有定时任务时才需启动）: celery -A celery_tasks.main beat
启动worker节点，运行任务: celery -A celery_tasks.main worker -l info
"""
celery的基本配置到此就完成了，使用下方的命令看一下能否成功启动：
celery -A celery_tasks.main worker -l info发布异步和延时任务
我们现在来实现一个发送短信验证码的异步任务（伪代码）
celery_tasks/_async/tasks.py
from celery_tasks.main import app
import time
# 调用 app实例对象的task方法, 装饰这个函数任务,可以设置name参数
@app.task(name='send_code')
def send_code(mobile, sms_code):
time.sleep(2)
print(f"短信发送中:手机号{mobile},验证码:{sms_code}")
接口调用 message/views.py
from django.http import JsonResponse
from celery_tasks._async.tasks import send_code
from datetime import datetime, timedelta
def async_send_code(request):
"""
异步处理
场景说明：用户注册时，需要发注册邮件和注册短信。
"""
# todo：开启异步短信发送
task_id = send_code.delay("157322285620", "123456")
# todo：开启延时任务
eta = datetime.utcnow() + timedelta(seconds=5)
task2_id = send_code.apply_async(args=("5s后发送", "12313"), eta=eta)
print(f"任务ID:{task_id}、{task2_id}")
return JsonResponse({"msg": "success"})
写完之后调用接口时，你会看到响应会立马返回，但是控制台会在2s后打印第一条信息，7s后打印第2条信息，因为我们在celery任务中设置了2s的睡眠，另外在接口函数中定义了一个5s后的定完成任务，至此异步和延时完成。
发布定时任务我们只需修改config.py文件即可
....
# 需要执行任务的配置
beat_schedule = {
"task1": {
"task": "send_code",
"schedule": 3.0,
"args": (1, 1) # # 任务函数参数
}
}
....
这里我们发布了一个定时任务task1，每3s执行一次send_code。
启动定时任务需要再控制台额外执行：
celery -A celery_tasks.main beat实现秒杀系统的消息队列
tasks.py
@app.task(name='order_kill')
def order_kill():
"""
-请求来到后端，提交一个celery任务---》celery任务异步的执行判断数量是否够，如果够，要生成订单（mysql）
-秒杀是否成功的结果还没有，直接返回了（返回任务id）
-前端启动一个定时任务，每隔5s，向后台发送一个查询请求，查询秒杀任务是否执行完成（带着任务id查）
-如果是未执行状态，或者执行中---》返回给前端，前端不处理，定时任务继续执行
-又隔了5s，发送查询，查询到秒杀成功的结果，返回给前端，秒杀成功
"""
print(f"订单秒杀中..")
# todo：数据库查询商品库存
comm = CommodityKill.objects.filter(id=1).first()
if not comm:
return False
# todo：有库存，要生成订单
if comm.count > 0:
comm.count -= 1
comm.save()
print(f"秒杀成功，剩余{comm.count}")
return True
# todo：库存为0，商品已被抢完
else:
print(f"商品库存不足...")
return False
views.py
def order_seconds_kill(request):
"""
订单秒杀
场景说明：所有秒杀订单进入消息队列，库存消耗完成之后秒杀失败
"""
task_id = order_kill.delay()
time.sleep(1)
task = AsyncResult(id=str(task_id), app=app)
if task.successful():
result = task.get()
if result:
print("购买成功")
else:
print("购买失败")
return JsonResponse({"msg": "success"})
作者：大家好，我是阿旭
支持我：点赞收藏⭐️留言
GitHub主页：Tengxu666 欢迎star
