Python定時任務(上)

Photo from Unsplash

 在項目中，我們可能遇到有定時任務的需求。其一：定時執行任務。例如每天早上 8 點定時推送早報。其二：每隔一個時間段就執行任務。比如：每隔一個小時提醒自己起來走動走動，避免長時間坐著。今天，我跟大家分享下 Python 定時任務的實現方法。

第一種辦法是最簡單又最暴力。那就是在一個死循環中，使用線程睡眠函數 sleep()。

from datetime import datetime
import time

'''
每個 10 秒列印當前時間。
'''
def timedTask():
   while True:
       print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
       time.sleep(10)

if __name__ == '__main__':
   timedTask()

這種方法能夠執行固定間隔時間的任務。如果 timedTask() 函數之後還有些操作，我們還使用死循環 + 阻塞線程。這會使得 timedTask() 一直佔有 CPU 資源，導致後續操作無法執行。我建議謹重使用。

既然第一種方法暴力，那麼有沒有比較優雅地方法？答案是肯定的。Python 標準庫 threading 中有個 Timer 類。它會新啟動一個線程來執行定時任務，所以它是非阻塞函式。

如果你有使用多線程的話，需要關心線程安全問題。那麼你可以選使用  threading.Timer 模塊。

from datetime import datetime
from threading import Timer
import time

'''
每個 10 秒列印當前時間。
'''
def timedTask():
   '''
   第一個參數: 延遲多長時間執行任務(單位: 秒)
   第二個參數: 要執行的任務, 即函數
   第三個參數: 調用函數的參數(tuple)
   '''
   Timer(10, task, ()).start()


def task():
   print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

if __name__ == '__main__':
   timedTask()
   while True:
       print(time.time())
       time.sleep(5)

運行結果：

第三種方式是使用標準庫中 sched 模塊。sched 是事件調度器，它通過  scheduler 類來調度事件，從而達到定時執行任務的效果。

sched 庫使用起來也是非常簡單。

1）首先構造一個 sched.scheduler 類

它接受兩個參數： timefunc 和  delayfunc。timefunc 應該返回一個數字，代表當前時間，delayfunc 函數接受一個參數，用於暫停運行的時間單元。

一般使用默認參數就行，即傳入這兩個參數 `time.time` 和 `time.sleep`.當然，你也可以自己實現時間暫停的函數。

2）添加調度任務

scheduler 提供了兩個添加調度任務的函數:

enter(delay, priority, action, argument=(), kwargs={})

該函數可以延遲一定時間執行任務。delay 表示延遲多長時間執行任務，單位是秒。priority 為優先級，越小優先級越大。兩個任務指定相同的延遲時間，優先級大的任務會向被執行。action 即需要執行的函數，argument  和  kwargs 分別是函數的位置和關鍵字參數。

scheduler.enterabs(time, priority, action, argument=(), kwargs={})

添加一項任務，但這個任務會在 time 這時刻執行。因此，time 是絕對時間.其他參數用法與 enter() 中的參數用法是一致。

3）把任務運行起來

調用 scheduler.run() 函數就完事了。

下面是 sche 使用的簡單示例：

from datetime import datetime
import sched
import time

'''
每個 10 秒列印當前時間。
'''
def timedTask():
   
   scheduler = sched.scheduler(time.time, time.sleep)
   
   scheduler.enter(10, 1, task)
   
   scheduler.run()


def task():
   print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

if __name__ == '__main__':
   timedTask()

值得注意的是： scheduler 中的每個調度任務只會工作一次，不會無限循環被調用。如果想重複執行同一任務， 需要重複添加調度任務即可。

作者：猴哥

公眾號：極客猴

愛好讀書，喜歡鑽研技術，夢想成為文藝青年的 boy。