1.利用 while True: + sleep() 實(shí)現(xiàn)定時(shí)任務(wù)

位于 time 模塊中的 sleep(secs) 函數(shù)，可以實(shí)現(xiàn)令當(dāng)前執(zhí)行的線程暫停 secs 秒后再繼續(xù)執(zhí)行。所謂暫停，即令當(dāng)前線程進(jìn)入阻塞狀態(tài)，當(dāng)達(dá)到 sleep() 函數(shù)規(guī)定的時(shí)間后，再由阻塞狀態(tài)轉(zhuǎn)為就緒狀態(tài)，等待 CPU 調(diào)度。

基于這樣的特性我們可以通過 while 死循環(huán)+sleep() 的方式實(shí)現(xiàn)簡(jiǎn)單的定時(shí)任務(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()

主要缺點(diǎn)：

• 只能設(shè)定間隔，不能指定具體的時(shí)間，比如每天早上 8:00
• sleep 是一個(gè)阻塞函數(shù)，也就是說 sleep 這一段時(shí)間，程序什么也不能操作。
  

2.使用 Timeloop 庫運(yùn)行定時(shí)任務(wù)

Timeloop[2] 是一個(gè)庫，可用于運(yùn)行多周期任務(wù)。這是一個(gè)簡(jiǎn)單的庫，它使用 decorator 模式在線程中運(yùn)行標(biāo)記函數(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())

3.利用 threading.Timer 實(shí)現(xiàn)定時(shí)任務(wù)

threading 模塊中的 Timer 是一個(gè)非阻塞函數(shù)，比 sleep 稍好一點(diǎn)，timer 最基本理解就是定時(shí)器，我們可以啟動(dòng)多個(gè)定時(shí)任務(wù)，這些定時(shí)器任務(wù)是異步執(zhí)行，所以不存在等待順序執(zhí)行問題。

Timer(interval, function, args=[ ], kwargs={ })

• interval: 指定的時(shí)間
• 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í)行一次

4.利用內(nèi)置模塊 sched 實(shí)現(xiàn)定時(shí)任務(wù)

sched 模塊實(shí)現(xiàn)了一個(gè)通用事件調(diào)度器，在調(diào)度器類使用一個(gè)延遲函數(shù)等待特定的時(shí)間，執(zhí)行任務(wù)。同時(shí)支持多線程應(yīng)用程序，在每個(gè)任務(wù)執(zhí)行后會(huì)立刻調(diào)用延時(shí)函數(shù)，以確保其他線程也能執(zhí)行。

class sched.scheduler(timefunc, delayfunc) 這個(gè)類定義了調(diào)度事件的通用接口，它需要外部傳入兩個(gè)參數(shù)，timefunc 是一個(gè)沒有參數(shù)的返回時(shí)間類型數(shù)字的函數(shù)（常用使用的如 time 模塊里面的 time)，delayfunc 應(yīng)該是一個(gè)需要一個(gè)參數(shù)來調(diào)用、與 timefunc 的輸出兼容、并且作用為延遲多個(gè)時(shí)間單位的函數(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 對(duì)象主要方法：

• enter(delay, priority, action, argument) ，安排一個(gè)事件來延遲 delay 個(gè)時(shí)間單位。
• cancel(event)：從隊(duì)列中刪除事件。如果事件不是當(dāng)前隊(duì)列中的事件，則該方法將跑出一個(gè) ValueError。
• run()：運(yùn)行所有預(yù)定的事件。這個(gè)函數(shù)將等待（使用傳遞給構(gòu)造函數(shù)的 delayfunc() 函數(shù)），然后執(zhí)行事件，直到不再有預(yù)定的事件。
  

個(gè)人點(diǎn)評(píng)：比 threading.Timer 更好，不需要循環(huán)調(diào)用。

到此這篇關(guān)于Python 4種實(shí)現(xiàn)定時(shí)任務(wù)的方案的文章就介紹到這了,更多相關(guān)Python 實(shí)現(xiàn)定時(shí)任務(wù)方案內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論