Python Celery实现订单超时自动取消的完整配置与代码
本文详解如何使用Python Celery结合Django框架实现订单超时自动取消功能。内容涵盖Celery项目目录结构搭建、任务注册、Django环境初始化、Redis消息队列配置以及Crontab定时调度器的设置。通过具体代码示例，展示如何编写检查订单状态的任务逻辑，并配置ORDER_TIMEOUT参数，确保未支付订单在指定时间后自动变更为取消状态。
Celery实现订单超时取消概述
本文分享使用Celery实现订单超时自动取消的具体代码实现，供开发者参考。
Celery官方文档中关于定时任务使用的说明
项目目录结构搭建
项目目录结构
我们需要新增一个任务目录，例如order：
celey_tasks/\n\n ├── config.py\n\n ├── __init__.py\n\n ├── main.py\n\n ├── order/\n\n ├── __init__.py\n\n └── tasks.py
Celery主程序与任务注册
在main.py中，注册任务目录【注意，接下来后面我们使用django的模型处理，所以必须对django的配置进行引入】
import os
from celery import Celery
# 1. 创建示例对象
app = Celery("luffy")
# 2. 加载配置
app.config_from_object("celery_tasks.config")
# 3. 注册任务[自动搜索并加载任务]
# 参数必须必须是一个列表，里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["celery_tasks.order"])
# 4. 在终端下面运行celery命令启动celery
# celery -A 主程序 worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info订单检查任务编写
接下来，在order任务目录下， 创建固定名字的任务文件tasks.py，代码：
from celery_tasks.main import app
@app.task(name="check_order")
def check_order():
print("检查订单是否过期!!!")定时任务调度配置
接下来，我们需要把这个任务设置定时任务，所以需要借助Celery本身提供的Crontab模块。
在配置文件中，对定时任务进行注册：
# 任务队列的链接地址
broker_url = 'redis://127.0.0.1:6379/15'
# 结果队列的链接地址
result_backend = 'redis://127.0.0.1:6379/14'
from celery.schedules import crontab
from .main import app
# 定时任务的调度列表，用于注册定时任务
app.conf.beat_schedule = {
# Executes every Monday morning at 7:30 a.m.
'check_order_outtime': {
# 本次调度的任务
'task': 'check_order', # 这里的任务名称必须先到main.py中注册
# 定时任务的调度周期
# 'schedule': crontab(minute=0, hour=0), # 每周凌晨00:00
'schedule': crontab(), # 每分钟
# 'args': (16, 16), # 注意：任务就是一个函数，所以如果有参数则需要传递
},
}Celery服务启动命令
接下来，我们就可以重启Celery并启用Celery的定时任务调度器
先在终端下，运行celery的定时任务程序，以下命令：
celery -A celery_tasks.main beat # ycelery.main 是celery的主应用文件然后再新建一个终端，运行以下命令，上面的命令必须先指定：
celery -A celery_tasks.main worker --loglevel=info定时任务
Django集成与订单超时逻辑实现
经过上面的测试以后，我们接下来只需改造上面的任务函数，用于判断修改订单是否超时。
要完成订单的任务功能，如果需要调用django框架的模型操作，那么必须针对django框架进行配置加载和初始化。\n\nmain.py，代码
import os
from celery import Celery
# 1. 创建示例对象
app = Celery("luffy")
# 把celery和django进行组合，识别和加载django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
# 在当前clery中启动django框架，对django框架进行进行初始化
import django
django.setup()
# 2. 加载配置
app.config_from_object("celery_tasks.config")
# 3. 注册任务[自动搜索并加载任务]
# 参数必须必须是一个列表，里面的每一个任务都是任务的路径名称
# app.autodiscover_tasks(["任务1","任务2"])
app.autodiscover_tasks(["celery_tasks.sms","celery_tasks.order"])
# 4. 在终端下面运行celery命令启动celery
# celery -A 主程序 worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info注意，因为在django中是有时区配置的，所以，我们在django框架配置中也要修改时区配置。
任务代码tasks.py的实现：
from celery_tasks.main import app
from orders.models import Order
from datetime import datetime
from django.conf import settings
@app.task(name="check_order")
def check_order():
# 查询出所有已经超时的订单
# 超时条件： 当前时间 > (订单生成时间 + 超时时间) =====>>>> (当前时间 - 超时时间) > 订单生成时间
now = datetime.now().timestamp()
timeout_number = now - settings.ORDER_TIMEOUT
timeout = datetime.fromtimestamp(timeout_number)
timeout_order_list = Order.objects.filter(order_status=0, created_time__lte=timeout)
for order in timeout_order_list:
order.order_status = 3
order.save()订单超时时间配置
配置文件，settings/dev.py，代码：
# 设置订单超时超时的时间[单位: 秒]
ORDER_TIMEOUT = 12 * 60 * 60以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
