本文實例為大家分享了celery實現(xiàn)訂單超時取消的具體代碼，供大家參考，具體內容如下
Celery官方文檔中關于定時任務使用的說明
項目目錄結構
我們需要新增一個任務目錄，例如order：
celey_tasks/
├── config.py
├── __init__.py
├── main.py
├── order/
├── __init__.py
└── tasks.py
在main.py中，注冊任務目錄【注意，接下來后面我們使用django的模型處理，所以必須對django的配置進行引入】
import os
from celery import Celery
# 1. 創(chuàng)建示例對象
app = Celery("luffy")
# 2. 加載配置
app.config_from_object("celery_tasks.config")
# 3. 注冊任務[自動搜索并加載任務]
# 參數(shù)必須必須是一個列表，里面的每一個任務都是任務的路徑名稱
# 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任務目錄下， 創(chuàng)建固定名字的任務文件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), # 注意：任務就是一個函數(shù)，所以如果有參數(shù)則需要傳遞
},
}
接下來，我們就可以重啟Celery并啟用Celery的定時任務調度器
先在終端下，運行celery的定時任務程序，以下命令：
celery -A celery_tasks.main beat # ycelery.main 是celery的主應用文件
然后再新建一個終端，運行以下命令，上面的命令必須先指定：
celery -A celery_tasks.main worker --loglevel=info
定時任務
經過上面的測試以后，我們接下來只需改造上面的任務函數(shù)，用于判斷修改訂單是否超時。
要完成訂單的任務功能，如果需要調用django框架的模型操作，那么必須針對django框架進行配置加載和初始化。
main.py，代碼
import os
from celery import Celery
# 1. 創(chuàng)建示例對象
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. 注冊任務[自動搜索并加載任務]
# 參數(shù)必須必須是一個列表，里面的每一個任務都是任務的路徑名稱
# 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中是有時區(qū)配置的，所以，我們在django框架配置中也要修改時區(qū)配置。
任務代碼tasks.py的實現(xiàn)：
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
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
您可能感興趣的文章:- python3中celery異步框架簡單使用+守護進程方式啟動
- Python Celery異步任務隊列使用方法解析
- python使用celery實現(xiàn)異步任務執(zhí)行的例子
- python celery分布式任務隊列的使用詳解
- Python環(huán)境下安裝使用異步任務隊列包Celery的基礎教程
- Python中celery的使用
