Hello,大家好,我是Alex~
今天我教大家4種利用python實現(xiàn)定時任務(wù)的方案,都不是很難哦
01 利用 while True: + sleep() 實現(xiàn)定時任務(wù)
位于 time 模塊中的 sleep(secs) 函數(shù),可以實現(xiàn)令當前執(zhí)行的線程暫停 secs 秒后再繼續(xù)執(zhí)行。所謂暫停,即令當前線程進入阻塞狀態(tài),當達到 sleep() 函數(shù)規(guī)定的時間后,再由阻塞狀態(tài)轉(zhuǎn)為就緒狀態(tài),等待 CPU 調(diào)度。
基于這樣的特性我們可以通過 while 死循環(huán)+sleep() 的方式實現(xiàn)簡單的定時任務(wù)。
代碼示例:
import?datetime
import?time
def?time_printer():
????now?=?datetime.datetime.now()
????ts?=?now.strftime('%Y-%m-%d?%H:%M:%S')
????print('do?func?time?:',?ts)
def?loop_monitor():
????while?True:
????????time_printer()
????????time.sleep(5)??#?暫停?5?秒
if?__name__?==?"__main__":
????loop_monitor()
主要缺點:
只能設(shè)定間隔,不能指定具體的時間,比如每天早上 8:00
sleep 是一個阻塞函數(shù),也就是說 sleep 這一段時間,程序什么也不能操作。
02 使用 Timeloop 庫運行定時任務(wù)
Timeloop[2]?是一個庫,可用于運行多周期任務(wù)。這是一個簡單的庫,它使用 decorator 模式在線程中運行標記函數(shù)。
示例代碼:
import?time
from?timeloop?import?Timeloop
from?datetime?import?timedelta
tl?=?Timeloop()
@tl.job(interval=timedelta(seconds=2))
def?sample_job_every_2s():
????print?"2s?job?current?time?:?{}".format(time.ctime())
@tl.job(interval=timedelta(seconds=5))
def?sample_job_every_5s():
????print?"5s?job?current?time?:?{}".format(time.ctime())
@tl.job(interval=timedelta(seconds=10))
def?sample_job_every_10s():
????print?"10s?job?current?time?:?{}".format(time.ctime())
03 利用 threading.Timer 實現(xiàn)定時任務(wù)
threading 模塊中的 Timer 是一個非阻塞函數(shù),比 sleep 稍好一點,timer 最基本理解就是定時器,我們可以啟動多個定時任務(wù),這些定時器任務(wù)是異步執(zhí)行,所以不存在等待順序執(zhí)行問題。
Timer(interval, function, args=[ ], kwargs={ })
interval: 指定的時間
function: 要執(zhí)行的方法
args/kwargs: 方法的參數(shù)
代碼示例:
import?datetime
from?threading?import?Timer
def?time_printer():
????now?=?datetime.datetime.now()
????ts?=?now.strftime('%Y-%m-%d?%H:%M:%S')
????print('do?func?time?:',?ts)
????loop_monitor()
def?loop_monitor():
????t?=?Timer(5,?time_printer)
????t.start()
if?__name__?==?"__main__":
????loop_monitor()
備注:Timer 只能執(zhí)行一次,這里需要循環(huán)調(diào)用,否則只能執(zhí)行一次
04 利用內(nèi)置模塊 sched 實現(xiàn)定時任務(wù)
sched 模塊實現(xiàn)了一個通用事件調(diào)度器,在調(diào)度器類使用一個延遲函數(shù)等待特定的時間,執(zhí)行任務(wù)。同時支持多線程應(yīng)用程序,在每個任務(wù)執(zhí)行后會立刻調(diào)用延時函數(shù),以確保其他線程也能執(zhí)行。
class sched.scheduler(timefunc, delayfunc) 這個類定義了調(diào)度事件的通用接口,它需要外部傳入兩個參數(shù),timefunc 是一個沒有參數(shù)的返回時間類型數(shù)字的函數(shù)(常用使用的如 time 模塊里面的 time),delayfunc 應(yīng)該是一個需要一個參數(shù)來調(diào)用、與 timefunc 的輸出兼容、并且作用為延遲多個時間單位的函數(shù)(常用的如 time 模塊的 sleep)。
代碼示例:
import?datetime
import?time
import?sched
def?time_printer():
????now?=?datetime.datetime.now()
????ts?=?now.strftime('%Y-%m-%d?%H:%M:%S')
????print('do?func?time?:',?ts)
????loop_monitor()
def?loop_monitor():
????s?=?sched.scheduler(time.time,?time.sleep)??#?生成調(diào)度器
????s.enter(5,?1,?time_printer,?())
????s.run()
if?__name__?==?"__main__":
????loop_monitor()
scheduler 對象主要方法:
enter(delay, priority, action, argument),安排一個事件來延遲 delay 個時間單位。
cancel(event):從隊列中刪除事件。如果事件不是當前隊列中的事件,則該方法將跑出一個 ValueError。
run():運行所有預定的事件。這個函數(shù)將等待(使用傳遞給構(gòu)造函數(shù)的 delayfunc() 函數(shù)),然后執(zhí)行事件,直到不再有預定的事件。
個人點評:比 threading.Timer 更好,不需要循環(huán)調(diào)用。
最后
小伙伴們?nèi)羰窍矚g這篇文章的,可以點個贊哦~,感謝你的支持!