【Django+Vue3 线上教育平台项目实战】Celery赋能：优化订单超时处理与自动化定时任务调度

在这里插入图片描述

文章目录

前言⭐✨💫🔥📖
一、Celery⭐
- 1.基本概念及介绍:✨
- 2.使用步骤💫
二、订单超时取消订单（Celery）🔥
- 具体实现流程📖

前言⭐✨💫🔥📖

在构建复杂的Web应用，如线上教育平台时，异步任务处理与定时任务调度成为提升性能和用户体验的关键。Celery 作为分布式任务队列框架，以其高效、灵活和可扩展性在Python项目中广泛应用。

本文将简述Celery的基本概念，并详细指导如何在Django项目中集成Celery，以实现订单超时自动取消的功能。同时，我们还将探讨如何设置定时任务，对成功和失败的订单进行统一处理，以优化业务流程和提高系统自动化水平。

一、Celery⭐

在这里插入图片描述

项目仓库：https://github.com/celery/celery/
官方文档：https://docs.celeryproject.org/en/latest/
中文文档（3.1）：http://docs.jinkan.org/docs/celery/getting-started/index.html

1.基本概念及介绍:✨

Celery是一个由Python编写的分布式任务队列和异步处理框架，它使得开发者能够轻松地在后台处理耗时任务，而不会阻塞主程序的执行。
Celery的设计基于生产者消费者模型，通过消息中间件（Broker）来协调任务的发布和执行。

Celery的核心组件包括：

1.任务（Tasks）：定义了要异步执行的功能。它们通常是一个普通的Python函数，通过装饰器标记为Celery任务。
2.工作者（Workers）：运行在后台的进程，不断地从消息队列中取出任务并执行。你可以运行多个工作者来并行处理任务。
3.消息中间件（Message Broker）：用于***接收任务并将它们放入队列中，同时负责结果的存储。Celery 支持多种中间件，如RabbitMQ、Redis、Amazon SQS等。
4.定时任务配制（beat）：三种模式 cron模式、date模式、interval模式
5.任务结果后端（Result Backend）：用于存储任务的结果，以便任务完成后可以查询。后端可以是数据库、缓存系统等，Celery支持包括Redis、RabbitMQ、SQLAlchemy在内的多种存储方式。

Celery的整体流程：

应用程序将一个任务添加到任务队列中。
消息中间件接收这个任务并将它存储在适当的消息队列里。
Celery 工作者监听这些队列，一旦发现新任务，就会取出并执行它。
任务执行的结果可以被存储在结果后端，供后续查询。

Celery 支持任务的链式执行、定时任务（使用celery beat）、任务优先级设置、任务结果的跟踪和重试机制等高级特性，非常适合需要高性能异步处理和任务调度的场景。

2.使用步骤💫

1.创建Celery：web_back/celery.py

import os
from celery import Celery

# 必须在实例化celery应用对象之前执行
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'web_back.settings')

# 实例化celery应用对象
app = Celery('web_back')
# 指定任务的队列名称
app.conf.task_default_queue = 'Celery'
# 也可以把配置写在django的项目配置中
app.config_from_object('django.conf:settings', namespace='CELERY') # 设置django中配置信息以 "CELERY_"开头为celery的配置信息
# 自动根据配置查找django的所有子应用下的tasks任务文件
app.autodiscover_tasks()

2.使用Celery：web_back/__init__.py

import pymysql
pymysql.version_info=(1,4,3,"final",0) # 指定了pymysql的版本：1.4.3,按照版本修改
pymysql.install_as_MySQLdb()

from .celery import app as celery_app

__all__ = ['celery_app']

3.settings.py 配置

# Celery配置
# from kombu import Exchange, Queue
# 设置任务接受的类型，默认是{'json'}
CELERY_ACCEPT_CONTENT = ['application/json']
# 设置task任务序列列化为json
CELERY_TASK_SERIALIZER = 'json'
# 请任务接受后存储时的类型
CELERY_RESULT_SERIALIZER = 'json'
# 时间格式化为中国时间
CELERY_TIMEZONE = 'Asia/Shanghai'
# 是否使用UTC时间
CELERY_ENABLE_UTC = False
# 指定borker为redis 如果指定rabbitmq CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
# 指定存储结果的地方，支持使用rpc、数据库、redis等等，具体可参考文档
# CELERY_RESULT_BACKEND = 'db+mysql://scott:tiger@localhost/foo'
# mysql 作为后端数据库
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'
# 设置任务过期时间 默认是一天，为None或0 表示永不过期
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 设置worker并发数，默认是cpu核心数
# CELERYD_CONCURRENCY = 12
# 设置每个worker最大任务数
CELERYD_MAX_TASKS_PER_CHILD = 100
# 指定任务的位置
CELERY_IMPORTS = (
    'base.tasks',
)
# 使用beat启动Celery定时任务
# schedule时间的具体设定参考：https://docs.celeryproject.org/en/stable/userguide/periodic-tasks.html

# 定时任务
CELERY_BEAT_SCHEDULE = {
    "orders_notpay": {  # 定时任务
        'task': 'base.tasks.cancelorders',
        'schedule': 3,  # 定时任务的调用时间，10表示每隔10秒调用一次任务，5*60表示每5分钟
        # "schedule": crontab(hour=7, minute=30, day_of_week=1),,  # 定时任务的调用时间，每周一早上7点30分调用一次add任务
    },
    "syn_mysqlEs": {  # 定时任务：每一分钟同步一次mysql到es的数据
        'task': 'base.tasks.syn_mysql_es',
        'schedule': 3,  # 定时任务的调用时间，10表示每隔10秒调用一次任务，5*60表示每5分钟
        # "schedule": crontab(hour=7, minute=30, day_of_week=1),,  # 定时任务的调用时间，每周一早上7点30分调用一次add任务
    },

}

4.设置tasks 任务：

#base/tasks.py
from celery import shared_task
from ronglianyunapi import send_sms as sms
# 记录日志：
import logging
logger = logging.getLogger("django")

@shared_task(name="send_sms")
def send_sms(tid, mobile, datas):
    """异步发送短信"""
    try:
        return sms(tid, mobile, datas)
    except Exception as e:
        logger.error(f"发送短信失败: {e}")

5.启动定时任务：

第一个终端：celery -A web_back worker -l INFO -P eventlet
第二个终端：celery -A web_back beat

二、订单超时取消订单（Celery）🔥

实现流程：

1.生成订单把当前时间+1800，和订单号存入队列中
2.在支付完的回调接口中，同步根据订单号查询订单，根据查询结果改变订单状态。异步从请求中获取状态，100000支付成功，队列中删除订单号
3.此时队列中的就是没支付的，1分钟执行一次，查询时间<当前时间，说明已经过期了，把订单改为失败状态，积分优惠券恢复

具体实现流程📖

1.生成订单接口 OrderView

2.在支付完的回调接口中，同步根据订单号查询订单，根据查询结果改变订单状态。异步从请求中获取状态，100000支付成功，队列中删除订单号

Celery定时任务：

订单超时取消订单
处理订单成功队列
处理订单失败队列

@shared_task
def cancelorders():
    # 获取订单号 0 小于等于int（time.time()）
    # 遍历订单号列表
    # 根据订单号查询订单表，把订单更新为失败状态
    # 查看订单是否使用积分和优惠券，如果使用更新，写入积分记录表，用户表中的积分+，优惠券+
    orderno = r.get_str("orderno")
    order = OrdersModel.objects.filter(transaction=orderno).first()
    pay_status = order.pay_status
    if pay_status == 1: #生成订单-未支付状态
        order.orders_status = 7 #订单-->取消状态
    #...
    print("取消订单操作")

@shared_task
def ordersuccess():
    #读取成功的队列，获取到订单号
    len = r.list_len("ordersuccess")
    while len > 0:
        sid = r.list_pop("ordersuccess")
        # 根据订单号查询订单
        order = OrdersDetailModel.objects.filter(orders_id=sid).first()
        # 更新用户课程表
        UserCourseModel.objects.create(user_id=order.user_id,course_id=order.course_id)
        # 更新用户课程章节表
        UserCourseChapterModel.objects.create(user_id=order.user_id,course_id=order.course_id,name=order.course.name)
        print("订单成功时，更新用户课程表和用户课程章节表")

@shared_task
def orderfail():
    # 读取失败的队列，获取到订单号，
    len = r.list_len("orderfail")
    while len > 0:
        fid = r.list_pop("orderfail")
        # 根据订单号查询订单
        order = OrdersModel.objects.filter(orders_id=fid).first()
        # 更新积分记录表、用户表中总积分、优惠券
        srecord = ScoreRecordModel.objects.filter(user_id=order.user_id).first()
        srecord.type = 1
        srecord.score = order.score
        srecord.save()
        user = UsersModel.objects.filter(user_id=order.user_id).first()
        user.points += order.score
        userCoupon = UserCouponModel.objects.filter(user_id=order.user_id).first()
        userCoupon.status = "未使用"
        print("订单失败时，更新数据完成")