Python-定時執行任務 - 肥皂的小屋

起因#

最近在寫自動簽到腳本，想掛在伺服器上每天 12 點執行，由此牽扯出一大堆類似的定時問題

本篇解答一下

每隔 xx 秒執行一次程序#

這是基礎，用time.sleep即可做到，直接上程序：

# 每5秒打印一次
from datetime import datetime
from time import sleep


def print_message():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("https://soapffz.com")


while True:
    print_message()
    sleep(5)

由此可以嘗試我們的目標：

from datetime import datetime
from time import sleep


def print_message():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("https://soapffz.com")


while True:
    while True:
        now = datetime.now()
        if now.hour == 0 and now.minute == 0:
            break
        sleep(20)
    print_message()

雖然簡單粗暴，但是這個while循環將會一直佔用CPU資源，一般不建議使用

threading.Timer 定時器#

上面一種太過於暴力，我們換一種優雅的方法

threading中有個Timer類。它會新啟動一個線程來執行定時任務，所以它是非阻塞函式。

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

Timer()函數可接受三個參數：

Timer(10, task, ()).start()
延遲多長時間執行任務(單位: 秒)
要執行的任務, 即函數
調用函數的參數(tuple)

延遲 xx 秒執行程序，只一次#

from datetime import datetime
from time import sleep
from threading import Timer


def print_message():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("https://soapffz.com")


Timer(10, print_message).start()

每隔 xx 秒執行一次程序，一直執行#

其實就是把Timer()語句移到執行的函數裡

from datetime import datetime
from time import sleep
from threading import Timer


def print_message():
    Timer(10, print_message).start()
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("https://soapffz.com")


print_message()

同理，需要每隔 24 小時 (不論起始時間)，只需要把Timer函數中的時間改為 86400 即可

schedule 模組定時執行任務#

官方文檔：https://schedule.readthedocs.io/en/stable/

這是一個輕量級的定時任務調度的庫，我們來看幾個例子：

from datetime import datetime
from time import sleep
import schedule


def print_message():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    print("https://soapffz.com")

schedule.every(15).seconds.do(print_message)
schedule.every().minutes.do(print_message)
schedule.every().hour.do(print_message)
schedule.every().day.at("10:30").do(print_message)
schedule.every(5).to(10).days.do(print_message)
schedule.every().wednesday.at("13:15").do(print_message)

while True:
    schedule.run_pending()
    sleep(1)

上面的意思就是：

每隔15秒執行一次任務
每隔1分鐘執行一次任務
每隔1小時執行一次任務
每天的10:30執行一次任務
每隔5到10天執行一次任務
每週一的這個時候執行一次任務
每週三13:15執行一次任務
run_pending：運行所有可以運行的任務

注意到運行結果中有一次重複打印了，不知道什麼原因～可以在留言區探討

schedule.run_pending () 是個什麼東西呢 ——

schedule 其實就只是個定時器。在 while True 死循環中，schedule.run_pending () 是保持 schedule 一直運行，去查詢上面那一堆的任務，在任務中，就可以設置不同的時間去運行。跟 linux 中設置 crontab 定時任務是類似的。

所以，schedule 有一定的局限性，所以只能用來執行一些小型的定時任務，它的局限性在哪呢 ——

1. 需要定時運行的函數 job 不應當是死循環類型的，也就是說，這個線程應該有一個執行完畢的出口。一是因為線程萬一僵死，會是非常棘手的問題；二是下一次定時任務還會開啟一個新的線程，執行次數多了就會演變成災難。

2. 如果 schedule 的時間間隔設置得比 job 執行的時間短，一樣會線程堆積形成災難，也就是說，我 job 的執行時間是 1 個小時，但是我定時任務設置的是 5 分鐘一次，那就會一直堆積線程。

另外還有一個庫sched.scheduler，用法和Timer()差不多，有興趣的可以自己了解一下

總結#

最後，我使用了datetime+sleep搭配的第一種方法和schedule的第三種方法做對比

伺服器配置：1 核心 CPU，1838MB 內存

第一種在平時的消耗如下：

第三種在平時的消耗如下：

雖然看起來後面一種好像平時更佔優勢，但是當時間到達指定時間的時候

前者很穩定，也不佔什麼 CPU，後者不知道發什麼瘋，佔了所有的內存和 CPU，給我推送了超多消息

但是我看了下程序也沒問題，所以我個人還是建議使用前者，稍微穩妥一點，雙重循環還是很致命的。

就這樣，本文完。

參考文章：