您好,登錄后才能下訂單哦!
本篇內(nèi)容主要講解“如何利用Python實(shí)現(xiàn)定時(shí)程序”,感興趣的朋友不妨來看看。本文介紹的方法操作簡(jiǎn)單快捷,實(shí)用性強(qiáng)。下面就讓小編來帶大家學(xué)習(xí)“如何利用Python實(shí)現(xiàn)定時(shí)程序”吧!
什么是定時(shí)器呢?它是指從指定的時(shí)刻開始,經(jīng)過一個(gè)指定時(shí)間,然后觸發(fā)一個(gè)事件,用戶可以自定義定時(shí)器的周期與頻率。
在 Python 中,如何定義一個(gè)定時(shí)器函數(shù)呢?我們先看第一種方法。假設(shè)我們需要執(zhí)行一個(gè)函數(shù)userCountFunc,這個(gè)函數(shù)需要每隔一個(gè)小時(shí)被執(zhí)行一次。那么,我們可以這樣寫:
def main(): startCronTask(userCountFunc, minutes=60) if __name__ == "__main__": main()
如上面的代碼,我們?cè)诙x了一個(gè) main 函數(shù)后,便定義了一個(gè)定時(shí)函數(shù) startCronTask。第一個(gè)參數(shù)為函數(shù)名,第二個(gè)參數(shù)為時(shí)間,第二個(gè)參數(shù)表示多長(zhǎng)時(shí)間后調(diào)用后面第一個(gè)參數(shù)的函數(shù)。第一個(gè)參數(shù)注意是函數(shù)對(duì)象,進(jìn)行參數(shù)傳遞,用函數(shù)名(如 userCountFunc)表示該對(duì)象,不是函數(shù)執(zhí)行語(yǔ)句 userCountFunc(),不然會(huì)報(bào)錯(cuò)。那么,在實(shí)現(xiàn)這個(gè)函數(shù)時(shí),需要引入定時(shí)功能,Python 中有一個(gè)定時(shí)任務(wù)模塊 BlockingScheduler:
from apscheduler.schedulers.blocking import BlockingScheduler def startCronTask(task, **config): # BlockingScheduler scheduler = BlockingScheduler() scheduler.add_job(task, "interval", **config) scheduler.start()
定義完一個(gè)調(diào)度模塊之后,實(shí)際的定時(shí)調(diào)度功能就完成了。接下來,需要具體實(shí)現(xiàn)定時(shí)執(zhí)行的邏輯函數(shù) userCountFunc:
def userCountFunc(): logger.info("count user") ...
這樣,對(duì)于方案一,實(shí)現(xiàn)的簡(jiǎn)單的定時(shí)功能就完成了。
方案一中介紹的是 Python 自帶的 BlockingScheduler 模塊,Python 中除了可以通過 BlockingScheduler,還通過線程實(shí)現(xiàn)定時(shí)器 timer,來簡(jiǎn)單的看下代碼:
import threading def timerFunc(): print("Hello World~") timer = threading.Timer(1, timerFunc) timer.start()
在上面的代碼中,定時(shí)器函數(shù) threading.Timer 主要有2個(gè)參數(shù),參數(shù)意義與方案一類似,接下來執(zhí)行這段程序:
Hello World~
Process finished with exit code 0
我們發(fā)現(xiàn)只執(zhí)行一遍,程序就結(jié)束了,但顯然不是我們想要的結(jié)果。其實(shí),我們看下 Time 類,有這樣的一句解釋性注釋:Call a function after a specified number of seconds,我們發(fā)現(xiàn)上面在執(zhí)行后并未循環(huán)執(zhí)行,所以需要修改下:
import threading def timerFunc(): print("Hello World~") global timer timer = threading.Timer(10.5, timerFunc) timer.start() timer = threading.Timer(3, timerFunc) timer.start()
此時(shí),我們可以看到輸出結(jié)果:
Hello World~
Hello World~
Hello World~
...
這里需要注意的是:必須在定時(shí)器執(zhí)行函數(shù)內(nèi)部重復(fù)構(gòu)造定時(shí)器,因?yàn)槎〞r(shí)器構(gòu)造后只執(zhí)行1次,必須循環(huán)調(diào)用。
另外,在上面的代碼中,我們其實(shí)還可以看到:threading.Timer(5.5, timerFunc),定時(shí)器間隔單位是秒,可以是浮點(diǎn)數(shù),如5.5,0.9等,在執(zhí)行函數(shù) timerFunc 內(nèi)部和外部中給的值可以不同。如上例中第一次執(zhí)行 timerFunc 是3秒后,后面的都是10.5秒后執(zhí)行。
接下來,我們?cè)倏纯慈绾卧僖欢〞r(shí)間結(jié)束定時(shí)功能。我們可以使用cancel停止定時(shí)器的工作,如下例:
import threading def timerFunc(): print("Hello World~") global timer timer = threading.Timer(10.5, timerFunc) timer.start() timer = threading.Timer(3, timerFunc) timer.start() time.sleep(60) timer.cancel()
上面的代碼表示:在定時(shí)器按照一定時(shí)間執(zhí)行后,執(zhí)行過程耗時(shí)60秒后停止定時(shí)操作功能,退出。顯示結(jié)果為:
Hello World~
Hello World~
Hello World~
Hello World~
Hello World~
...Process finished with exit code 0
到此,相信大家對(duì)“如何利用Python實(shí)現(xiàn)定時(shí)程序”有了更深的了解,不妨來實(shí)際操作一番吧!這里是億速云網(wǎng)站,更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢,關(guān)注我們,繼續(xù)學(xué)習(xí)!
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。