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

位于 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)簡單的定時任務。

代碼示例：

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()

主要缺點：

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

2.使用 Timeloop 庫運行定時任務

Timeloop[2] 是一個庫，可用于運行多周期任務。這是一個簡單的庫，它使用 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())

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

threading 模塊中的 Timer 是一個非阻塞函數(shù)，比 sleep 稍好一點，timer 最基本理解就是定時器，我們可以啟動多個定時任務，這些定時器任務是異步執(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í)行一次

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

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

class sched.scheduler(timefunc, delayfunc) 這個類定義了調(diào)度事件的通用接口，它需要外部傳入兩個參數(shù)，timefunc 是一個沒有參數(shù)的返回時間類型數(shù)字的函數(shù)（常用使用的如 time 模塊里面的 time)，delayfunc 應該是一個需要一個參數(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ù)將等待（使用傳遞給構造函數(shù)的 delayfunc() 函數(shù)），然后執(zhí)行事件，直到不再有預定的事件。
  

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

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

最新評論