在項目中,我們可能遇到有定時任務的需求。其一:定時執(zhí)行任務赡盘。例如每天早上 8 點定時推送早報。其二:每隔一個時間段就執(zhí)行任務缰揪。比如:每隔一個小時提醒自己起來走動走動陨享,避免長時間坐著。今天钝腺,我跟大家分享下 Python 定時任務的實現方法抛姑。
1
第一種辦法是最簡單又最暴力。那就是在一個死循環(huán)中艳狐,使用線程睡眠函數 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()
這種方法能夠執(zhí)行固定間隔時間的任務毫目。如果timedTask()
函數之后還有些操作蔬啡,我們還使用死循環(huán) + 阻塞線程诲侮。這會使得timedTask()
一直占有 CPU 資源,導致后續(xù)操作無法執(zhí)行箱蟆。我建議謹重使用沟绪。
2
既然第一種方法暴力,那么有沒有比較優(yōu)雅地方法空猜?答案是肯定的绽慈。Python 標準庫 threading 中有個 Timer 類。它會新啟動一個線程來執(zhí)行定時任務辈毯,所以它是非阻塞函式坝疼。
如果你有使用多線程的話,需要關心線程安全問題谆沃。那么你可以選使用threading.Timer
模塊钝凶。
from datetime import datetime
from threading import Timer
import time
'''
每個 10 秒打印當前時間。
'''
def timedTask():
'''
第一個參數: 延遲多長時間執(zhí)行任務(單位: 秒)
第二個參數: 要執(zhí)行的任務, 即函數
第三個參數: 調用函數的參數(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)
運行結果:
1512486945.1196375
1512486950.119873
2017-12-05 23:15:50
1512486955.133385
3
第三種方式是使用標準庫中sched
模塊唁影。sched 是事件調度器腿椎,它通過 scheduler
類來調度事件,從而達到定時執(zhí)行任務的效果夭咬。
sched
庫使用起來也是非常簡單。
1)首先構造一個sched.scheduler
類
它接受兩個參數:timefunc
和 delayfunc
铆隘。timefunc 應該返回一個數字卓舵,代表當前時間,delayfunc 函數接受一個參數膀钠,用于暫停運行的時間單元掏湾。
一般使用默認參數就行,即傳入這兩個參數 time.time
和 time.sleep
.當然肿嘲,你也可以自己實現時間暫停的函數融击。
2)添加調度任務
scheduler
提供了兩個添加調度任務的函數:
enter(delay, priority, action, argument=(), kwargs={})
該函數可以延遲一定時間執(zhí)行任務。delay
表示延遲多長時間執(zhí)行任務雳窟,單位是秒尊浪。priority
為優(yōu)先級,越小優(yōu)先級越大封救。兩個任務指定相同的延遲時間拇涤,優(yōu)先級大的任務會向被執(zhí)行。action
即需要執(zhí)行的函數誉结,argument
和 kwargs
分別是函數的位置和關鍵字參數鹅士。
scheduler.enterabs(time, priority, action, argument=(), kwargs={})
添加一項任務,但這個任務會在 time
這時刻執(zhí)行惩坑。因此掉盅,time
是絕對時間.其他參數用法與 enter()
中的參數用法是一致也拜。
3)把任務運行起來
調用 scheduler.run()
函數就完事了。
下面是 sche 使用的簡單示例:
from datetime import datetime
import sched
import time
'''
每個 10 秒打印當前時間趾痘。
'''
def timedTask():
# 初始化 sched 模塊的 scheduler 類
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 中的每個調度任務只會工作一次慢哈,不會無限循環(huán)被調用。如果想重復執(zhí)行同一任務扼脐, 需要重復添加調度任務即可岸军。
上篇閱讀:Python 繪圖,我只用 Matplotlib(三)—— 柱狀圖
推薦閱讀:徹底理解Iterable瓦侮、Iterator艰赞、generator