python使用celery實(shí)現(xiàn)訂單超時(shí)取消
本文實(shí)例為大家分享了celery實(shí)現(xiàn)訂單超時(shí)取消的具體代碼，供大家參考，具體內(nèi)容如下
Celery官方文檔中關(guān)于定時(shí)任務(wù)使用的說(shuō)明
項(xiàng)目目錄結(jié)構(gòu)
我們需要新增一個(gè)任務(wù)目錄，例如order：
celey_tasks/
├── config.py
├── __init__.py
├── main.py
├── order/
├── __init__.py
└── tasks.py
在main.py中，注冊(cè)任務(wù)目錄【注意，接下來(lái)后面我們使用django的模型處理，所以必須對(duì)django的配置進(jìn)行引入】
import os
from celery import Celery
# 1. 創(chuàng)建示例對(duì)象
app = Celery("luffy")
# 2. 加載配置
app.config_from_object("celery_tasks.config")
# 3. 注冊(cè)任務(wù)[自動(dòng)搜索并加載任務(wù)]
# 參數(shù)必須必須是一個(gè)列表，里面的每一個(gè)任務(wù)都是任務(wù)的路徑名稱
# app.autodiscover_tasks(["任務(wù)1","任務(wù)2"])
app.autodiscover_tasks(["celery_tasks.order"])
# 4. 在終端下面運(yùn)行celery命令啟動(dòng)celery
# celery -A 主程序 worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info
接下來(lái)，在order任務(wù)目錄下， 創(chuàng)建固定名字的任務(wù)文件tasks.py，代碼：
from celery_tasks.main import app
@app.task(name="check_order")
def check_order():
print("檢查訂單是否過(guò)期!!!")
接下來(lái)，我們需要把這個(gè)任務(wù)設(shè)置定時(shí)任務(wù)，所以需要借助Celery本身提供的Crontab模塊。
在配置文件中，對(duì)定時(shí)任務(wù)進(jìn)行注冊(cè)：
# 任務(wù)隊(duì)列的鏈接地址
broker_url = 'redis://127.0.0.1:6379/15'
# 結(jié)果隊(duì)列的鏈接地址
result_backend = 'redis://127.0.0.1:6379/14'
from celery.schedules import crontab
from .main import app
# 定時(shí)任務(wù)的調(diào)度列表，用于注冊(cè)定時(shí)任務(wù)
app.conf.beat_schedule = {
# Executes every Monday morning at 7:30 a.m.
'check_order_outtime': {
# 本次調(diào)度的任務(wù)
'task': 'check_order', # 這里的任務(wù)名稱必須先到main.py中注冊(cè)
# 定時(shí)任務(wù)的調(diào)度周期
# 'schedule': crontab(minute=0, hour=0), # 每周凌晨00:00
'schedule': crontab(), # 每分鐘
# 'args': (16, 16), # 注意：任務(wù)就是一個(gè)函數(shù)，所以如果有參數(shù)則需要傳遞
},
}
接下來(lái)，我們就可以重啟Celery并啟用Celery的定時(shí)任務(wù)調(diào)度器
先在終端下，運(yùn)行celery的定時(shí)任務(wù)程序，以下命令：
celery -A celery_tasks.main beat # ycelery.main 是celery的主應(yīng)用文件
然后再新建一個(gè)終端，運(yùn)行以下命令，上面的命令必須先指定：
celery -A celery_tasks.main worker --loglevel=info
定時(shí)任務(wù)
經(jīng)過(guò)上面的測(cè)試以后，我們接下來(lái)只需改造上面的任務(wù)函數(shù)，用于判斷修改訂單是否超時(shí)。
要完成訂單的任務(wù)功能，如果需要調(diào)用django框架的模型操作，那么必須針對(duì)django框架進(jìn)行配置加載和初始化。
main.py，代碼
import os
from celery import Celery
# 1. 創(chuàng)建示例對(duì)象
app = Celery("luffy")
# 把celery和django進(jìn)行組合，識(shí)別和加載django的配置文件
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
# 在當(dāng)前clery中啟動(dòng)django框架，對(duì)django框架進(jìn)行進(jìn)行初始化
import django
django.setup()
# 2. 加載配置
app.config_from_object("celery_tasks.config")
# 3. 注冊(cè)任務(wù)[自動(dòng)搜索并加載任務(wù)]
# 參數(shù)必須必須是一個(gè)列表，里面的每一個(gè)任務(wù)都是任務(wù)的路徑名稱
# app.autodiscover_tasks(["任務(wù)1","任務(wù)2"])
app.autodiscover_tasks(["celery_tasks.sms","celery_tasks.order"])
# 4. 在終端下面運(yùn)行celery命令啟動(dòng)celery
# celery -A 主程序 worker --loglevel=info
# celery -A celery_tasks.main worker --loglevel=info
注意，因?yàn)樵赿jango中是有時(shí)區(qū)配置的，所以，我們?cè)赿jango框架配置中也要修改時(shí)區(qū)配置。
任務(wù)代碼tasks.py的實(shí)現(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(): # 查詢出所有已經(jīng)超時(shí)的訂單 # 超時(shí)條件： 當(dāng)前時(shí)間 > (訂單生成時(shí)間 + 超時(shí)時(shí)間) =====>>>> (當(dāng)前時(shí)間 - 超時(shí)時(shí)間) > 訂單生成時(shí)間 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，代碼：
# 設(shè)置訂單超時(shí)超時(shí)的時(shí)間[單位: 秒] ORDER_TIMEOUT = 12 * 60 * 60
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
- python庫(kù)Celery異步發(fā)送電子郵件定時(shí)生成報(bào)告實(shí)戰(zhàn)示例
- python分布式庫(kù)celery處理大規(guī)模的任務(wù)并行化
- python分布式系統(tǒng)Celery安裝使用實(shí)例講解
- Python?Celery定時(shí)任務(wù)詳細(xì)講解
- Python中celery的使用
- 詳解Python Celery和RabbitMQ實(shí)戰(zhàn)教程
- Python celery原理及運(yùn)行流程解析
- python使用Celery構(gòu)建異步任務(wù)隊(duì)列提高服務(wù)器吞吐量及響應(yīng)速度
相關(guān)文章
Python使用Flask框架實(shí)現(xiàn)文件上傳實(shí)例
這篇文章主要介紹了Python使用Flask庫(kù)文件上傳實(shí)例,用?Flask?處理文件上傳很容易，只要確保HTML表單中設(shè)置enctype="multipart/form-data"屬性就可以了,需要的朋友可以參考下2023-08-08
Python的venv虛擬環(huán)境使用及說(shuō)明
這篇文章主要介紹了Python的venv虛擬環(huán)境使用及說(shuō)明,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2025-03-03
python根據(jù)距離和時(shí)長(zhǎng)計(jì)算配速示例
這篇文章主要介紹了python根據(jù)距離和時(shí)長(zhǎng)計(jì)算配速示例,需要的朋友可以參考下2014-02-02
python 實(shí)現(xiàn)一個(gè)貼吧圖片爬蟲(chóng)的示例
下面小編就為大家?guī)?lái)一篇python 實(shí)現(xiàn)一個(gè)貼吧圖片爬蟲(chóng)的示例。小編覺(jué)得挺不錯(cuò)的，現(xiàn)在就分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-10-10
python查找目錄下指定擴(kuò)展名的文件實(shí)例
這篇文章主要介紹了python查找目錄下指定擴(kuò)展名的文件,實(shí)例分析了Python文件查詢的技巧,非常具有實(shí)用價(jià)值,需要的朋友可以參考下2015-04-04
在本篇文章里小編給大家整理了一篇關(guān)于python超參數(shù)優(yōu)化的具體方法，有需要的朋友們可以學(xué)習(xí)下。2021-08-08
這篇文章主要介紹了pandas如何修改特定的值問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-02-02
Python基于pygame實(shí)現(xiàn)圖片代替鼠標(biāo)移動(dòng)效果
這篇文章主要介紹了Python基于pygame實(shí)現(xiàn)圖片代替鼠標(biāo)移動(dòng)效果,可實(shí)現(xiàn)將鼠標(biāo)箭頭轉(zhuǎn)換成圖形的功能,涉及pygame圖形操作的相關(guān)技巧,具有一定參考借鑒價(jià)值,需要的朋友可以參考下2015-11-11
這篇文章主要介紹了Python中os模塊功能與用法,總結(jié)分析了Python os模塊基本功能、內(nèi)置函數(shù)、使用方法及相關(guān)操作注意事項(xiàng),需要的朋友可以參考下2020-02-02
Python利用多線程同步鎖實(shí)現(xiàn)多窗口訂票系統(tǒng)(推薦)
這篇文章主要介紹了Python利用多線程同步鎖實(shí)現(xiàn)多窗口訂票系統(tǒng),主要是利用threading.lock（）通過(guò)實(shí)例代碼相結(jié)合給大家講解的非常詳細(xì)，具有一定的參考借鑒價(jià)值，需要的朋友可以參考下2019-12-12