Python定時(shí)任務(wù)工具之APScheduler使用方式
APScheduler (advanceded python scheduler)是一款Python開發(fā)的定時(shí)任務(wù)工具。
文檔地址 apscheduler.readthedocs.io/en/latest/u…
特點(diǎn):
- 不依賴于Linux系統(tǒng)的crontab系統(tǒng)定時(shí),獨(dú)立運(yùn)行
- 可以 動(dòng)態(tài)添加 新的定時(shí)任務(wù),如下單后30分鐘內(nèi)必須支付,否則取消訂單,就可以借助此工具(每下一單就要添加此訂單的定時(shí)任務(wù))
- 對(duì)添加的定時(shí)任務(wù)可以做持久保存
1 安裝
pip install apscheduler
2 組成
- APScheduler 由以下四部分組成:
- triggers 觸發(fā)器 指定定時(shí)任務(wù)執(zhí)行的時(shí)機(jī)
- job stores 存儲(chǔ)器 可以將定時(shí)持久存儲(chǔ)
- executors 執(zhí)行器 在定時(shí)任務(wù)該執(zhí)行時(shí),以進(jìn)程或線程方式執(zhí)行任務(wù)
- schedulers 調(diào)度器 常用的有BackgroundScheduler( 后臺(tái)運(yùn)行 )和BlockingScheduler( 阻塞式 )
3 使用方式
from apscheduler.schedulers.background import BlockingScheduler # 創(chuàng)建定時(shí)任務(wù)的調(diào)度器對(duì)象 scheduler = BlockingScheduler() # 創(chuàng)建執(zhí)行器 executors = { 'default': ThreadPoolExecutor(20), } # 定義定時(shí)任務(wù) def my_job(param1, param2): # 參數(shù)通過add_job()args傳遞傳遞過來 print(param1) # 100 print(param2) # python # 向調(diào)度器中添加定時(shí)任務(wù) scheduler.add_job(my_job, 'date', args=[100, 'python'], executors=executors) # 啟動(dòng)定時(shí)任務(wù)調(diào)度器工作 scheduler.start()
4 調(diào)度器 Scheduler
負(fù)責(zé)管理定時(shí)任務(wù)
BlockingScheduler : 作為獨(dú)立進(jìn)程時(shí)使用
from apscheduler.schedulers.blocking import BlockingScheduler scheduler = BlockingScheduler() scheduler.start() # 此處程序會(huì)發(fā)生阻塞 BackgroundScheduler : 在框架程序(如Django、Flask)中使用. from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() scheduler.start() # 此處程序不會(huì)發(fā)生阻塞
- AsyncIOScheduler : 當(dāng)你的程序使用了asyncio的時(shí)候使用。
- GeventScheduler : 當(dāng)你的程序使用了gevent的時(shí)候使用。
- TornadoScheduler : 當(dāng)你的程序基于Tornado的時(shí)候使用。
- TwistedScheduler : 當(dāng)你的程序使用了Twisted的時(shí)候使用
- QtScheduler : 如果你的應(yīng)用是一個(gè)Qt應(yīng)用的時(shí)候可以使用。
4 執(zhí)行器 executors
在定時(shí)任務(wù)該執(zhí)行時(shí),以進(jìn)程或線程方式執(zhí)行任務(wù)
ThreadPoolExecutor from apscheduler.executors.pool import ThreadPoolExecutor ThreadPoolExecutor(max_workers)
使用方法
from apscheduler.executors.pool import ThreadPoolExecutor executors = { 'default': ThreadPoolExecutor(20) # 最多20個(gè)線程同時(shí)執(zhí)行 } scheduler = BackgroundScheduler(executors=executors) ProcessPoolExecutor from apscheduler.executors.pool import ProcessPoolExecutor ProcessPoolExecutor(max_workers)
使用方法
from apscheduler.executors.pool import ProcessPoolExecutor executors = { 'default': ProcessPoolExecutor(5) # 最多5個(gè)進(jìn)程同時(shí)執(zhí)行 } scheduler = BackgroundScheduler(executors=executors)
5 觸發(fā)器 Trigger
指定定時(shí)任務(wù)執(zhí)行的時(shí)機(jī)。
1) date 在特定的時(shí)間日期執(zhí)行
from datetime import date # 在2019年11月6日00:00:00執(zhí)行 sched.add_job(my_job, 'date', run_date=date(2019, 11, 6)) # 在2019年11月6日16:30:05 sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5)) sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05') # 立即執(zhí)行 sched.add_job(my_job, 'date') sched.start()
2) interval 經(jīng)過指定的時(shí)間間隔執(zhí)行
weeks (int) – number of weeks to wait days (int) – number of days to wait hours (int) – number of hours to wait minutes (int) – number of minutes to wait seconds (int) – number of seconds to wait start_date (datetime|str) – starting point for the interval calculation end_date (datetime|str) – latest possible date/time to trigger on timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations from datetime import datetime # 每?jī)尚r(shí)執(zhí)行一次 sched.add_job(job_function, 'interval', hours=2) # 在2012年10月10日09:30:00 到2014年6月15日11:00:00的時(shí)間內(nèi),每?jī)尚r(shí)執(zhí)行一次 sched.add_job(job_function, 'interval', hours=2, start_date='2012-10-10 09:30:00', end_date='2014-06-15 11:00:00')
3) cron 按指定的周期執(zhí)行
year (int|str) – 4-digit year month (int|str) – month (1-12) day (int|str) – day of the (1-31) week (int|str) – ISO week (1-53) day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun) hour (int|str) – hour (0-23) minute (int|str) – minute (0-59) second (int|str) – second (0-59) start_date (datetime|str) – earliest possible date/time to trigger on (inclusive) end_date (datetime|str) – latest possible date/time to trigger on (inclusive) timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone) # 在6、7、8、11、12月的第三個(gè)周五的00:00, 01:00, 02:00和03:00 執(zhí)行 sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3') # 在2014年5月30日前的周一到周五的5:30執(zhí)行 sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
6.任務(wù)存儲(chǔ)
MemoryJobStore 默認(rèn)內(nèi)存存儲(chǔ) MongoDBJobStore 任務(wù)保存到MongoDB from apscheduler.jobstores.mongodb import MongoDB JobStoreMongoDBJobStore()復(fù)制代碼 RedisJobStore 任務(wù)保存到redis from apscheduler.jobstores.redis import RedisJobStore RedisJobStore()
7 配置方法
方法1
from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.executors.pool import ThreadPoolExecutor executors = { 'default': ThreadPoolExecutor(20), } conf = { # redis配置 "host":127.0.0.1, "port":6379, "db":15, # 連接15號(hào)數(shù)據(jù)庫 "max_connections":10 # redis最大支持300個(gè)連接數(shù) } scheduler = BackgroundScheduler(executors=executors) scheduler.add_jobstore(jobstore='redis', **conf) # 添加任務(wù)持久化存儲(chǔ)方式,如果未安裝redis可省略此步驟
方法2
from pytz import utc from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.executors.pool import ProcessPoolExecutor executors = { 'default': {'type': 'threadpool', 'max_workers': 20}, 'processpool': ProcessPoolExecutor(max_workers=5) } scheduler = BackgroundScheduler() # .. 此處可以編寫其他代碼 # 使用configure方法進(jìn)行配置 scheduler.configure(executors=executors)
8 啟動(dòng)
scheduler.start()
對(duì)于BlockingScheduler ,程序會(huì)阻塞在這,防止退出,作為獨(dú)立進(jìn)程時(shí)使用。(可以用來生成靜態(tài)頁面)
對(duì)于BackgroundScheduler,可以在應(yīng)用程序中使用。不再以單獨(dú)的進(jìn)程使用。(如30分鐘內(nèi)取消訂單)
9 擴(kuò)展
任務(wù)管理
方式1
job = scheduler.add_job(myfunc, 'interval', minutes=2) # 添加任務(wù) job.remove() # 刪除任務(wù) job.pause() # 暫定任務(wù) job.resume() # 恢復(fù)任務(wù)
方式2
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') # 添加任務(wù) scheduler.remove_job('my_job_id') # 刪除任務(wù) scheduler.pause_job('my_job_id') # 暫定任務(wù) scheduler.resume_job('my_job_id') # 恢復(fù)任務(wù)
調(diào)整任務(wù)調(diào)度周期
job.modify(max_instances=6, name='Alternate name') scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')復(fù)制代碼 停止APScheduler運(yùn)行 scheduler.shutdown()
10 綜合使用
這里提供30分鐘取消訂單支付的思路,可以使用Flask或者Django程序都能實(shí)現(xiàn),這里是在django應(yīng)用中動(dòng)態(tài)的添加一個(gè)定時(shí)任務(wù),調(diào)度器需要使用BackgroundScheduler。下面先定義執(zhí)行訂單取消的任務(wù)。
from apscheduler.executors.pool import ThreadPoolExecutor from datetime import datetime, timedelta from apscheduler.schedulers.blocking import BackgroundScheduler from goods.models import SKU from orders.models import OrderGoods def cancel_order_job(order_id, sku_id, stock, sales): # 將訂單商品和訂單信息篩選出來 order_goods = OrderGoods.objects.filter( order_id=order_id, sku_id=sku_id) order_goods.delete() # 刪除訂單 try: sku = SKU.objects.get(id=sku_id) sku.stock += stock # 訂單刪掉后商品表里的庫存恢復(fù) sku.sales -= sales # 商品表里銷量還原 sku.save() except Exception as e: print(e)
具體操作哪些表要根據(jù)自身表的設(shè)計(jì)來定,大致是上面的思路。然后在生成訂單的視圖中同時(shí)生成取消訂單的任務(wù)。然后將取消訂單cancel_order_job()
需要的參數(shù)傳遞過去,注意要判定當(dāng)前訂單的狀態(tài)為未支付狀態(tài)。
from datetime import datetime, timedelta class OrderCommitView(View): def post(self, request): # ... 此處省略生成訂單相關(guān)邏輯 if status == OrderInfo.STATUS.UNPADED: # 待支付狀態(tài) executors = { 'default': ThreadPoolExecutor(10) } now = datetime.now() delay = now + timedelta(minutes=30) # 從當(dāng)前下訂單延時(shí)30分鐘后 scheduler = BackgroundScheduler(executors=executors) # 添加定時(shí)任務(wù) scheduler.add_job(cancel_order_job, 'date', run_date=delay, args=[order_id, sku.id, sku.stock, sku.sales]) scheduler.start() # ....省略其他業(yè)務(wù)及返回
注意: 如果需要周期性的執(zhí)行一個(gè)定時(shí)任務(wù),如果用到了django中模型類或者Flask的配置信息等相關(guān)信息,需要將框架的配置信息導(dǎo)入。
總結(jié)
以上所述是小編給大家介紹的Python定時(shí)任務(wù)工具之APScheduler詳解,希望對(duì)大家有所幫助,如果大家有任何疑問請(qǐng)給我留言,小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持!
如果你覺得本文對(duì)你有幫助,歡迎轉(zhuǎn)載,煩請(qǐng)注明出處,謝謝!
相關(guān)文章
使用sklearn的cross_val_score進(jìn)行交叉驗(yàn)證實(shí)例
今天小編就為大家分享一篇使用sklearn的cross_val_score進(jìn)行交叉驗(yàn)證實(shí)例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2020-02-02Python實(shí)現(xiàn)帶GUI界面的手寫數(shù)字識(shí)別
這篇文章主要介紹了如何通過Python實(shí)現(xiàn)帶GUI界面的手寫數(shù)字識(shí)別,文中的示例代碼講解詳細(xì),對(duì)我們學(xué)習(xí)Python有一定的幫助,感興趣的可以了解一下2022-01-01python通過floor函數(shù)舍棄小數(shù)位的方法
這篇文章主要介紹了python通過floor函數(shù)舍棄小數(shù)位的方法,實(shí)例分析了Python中floor函數(shù)的功能及使用技巧,需要的朋友可以參考下2015-03-03Python 時(shí)間操作例子和時(shí)間格式化參數(shù)小結(jié)
這篇文章主要介紹了Python 時(shí)間操作例子,例如取前幾天、后幾天、前一月、后一月等,需要的朋友可以參考下2014-04-04python之pymysql模塊簡(jiǎn)單應(yīng)用示例代碼
這篇文章主要介紹了python之pymysql模塊簡(jiǎn)單應(yīng)用示例代碼,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-12-12python 實(shí)現(xiàn)在Excel末尾增加新行
下面小編就為大家分享一篇python 實(shí)現(xiàn)在Excel末尾增加新行,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過來看看吧2018-05-05詳解Python之?dāng)?shù)據(jù)序列化(json、pickle、shelve)
本篇文章主要介紹了Python之?dāng)?shù)據(jù)序列化,本節(jié)要介紹的就是Python內(nèi)置的幾個(gè)用于進(jìn)行數(shù)據(jù)序列化的模塊,有興趣的可以了解一下。2017-03-03python實(shí)現(xiàn)簡(jiǎn)單圖書管理系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)簡(jiǎn)單圖書管理系統(tǒng),文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下2019-11-11