在 Python 中實現「指定時間執行一次」或「定期重複執行」任務,有多種方案。本文對比三種常見做法:標準庫 sched、threading.Timer、以及第三方強大庫 APScheduler,幫助你根據場景選擇最適合的工具。
sched.scheduler(標準庫、單執行緒、阻塞)import sched, time, os
def play_music(path):
os.system(f'mpg123 "{path}"') # 或其他播放指令
def run_once_at(target_time: tuple, func, *args):
"""
target_time: (年, 月, 日, 時, 分, 秒)
"""
s = sched.scheduler(time.time, time.sleep)
run_ts = time.mktime(target_time + (0, 0, 0)) # 補齊 9 元組
delay = max(0, run_ts - time.time())
s.enter(delay, 1, func, args)
print(f'排程於 {time.ctime(run_ts)} 執行,延遲 {delay:.0f} 秒')
s.run() # 阻塞直到任務完成
if __name__ == '__main__':
# 例:今天 22:30 播放音樂
now = time.localtime()
target = (now.tm_year, now.tm_mon, now.tm_mday, 22, 30, 0)
run_once_at(target, play_music, '/home/user/music/alarm.mp3')
優點:無依賴、邏輯簡單。
缺點:run() 阻塞主執行緒、無持久化、重啟後排程遺失、難以管理多任務。
threading.Timer(標準庫、非阻塞、一次性)
import threading, time, os
def play_music(path):
os.system(f'mpg123 "{path}"')
def schedule_once(target_ts: float, func, *args):
delay = max(0, target_ts - time.time())
t = threading.Timer(delay, func, args)
t.daemon = True # 主程式結束時自動結束
t.start()
return t
if __name__ == '__main__':
target = time.mktime((2024, 12, 31, 23, 59, 50, 0, 0, 0))
timer = schedule_once(target, play_music, '/home/user/music/newyear.mp3')
print('主程式繼續執行其他工作...')
# 主執行緒可做別的事,Timer 在背景等待
優點:非阻塞、無額外依賴。
缺點:僅支援一次性任務、無重複排程、無管理介面、進程重啟失效。
安裝:pip install apscheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
import os, time
def job(msg):
print(f'[{time.strftime("%H:%M:%S")}] {msg}')
scheduler = BackgroundScheduler(timezone='Asia/Taipei')
scheduler.start()
# 1. 固定間隔(每 10 秒)
scheduler.add_job(job, 'interval', seconds=10, args=['間隔任務'])
# 2. Cron 風格(每天 09:00)
scheduler.add_job(job, CronTrigger(hour=9, minute=0), args=['每日早安'])
# 3. 指定單一時間點
run_date = '2024-12-31 23:59:50'
scheduler.add_job(job, DateTrigger(run_date=run_date), args=['跨年倒數'])
# 4. 持久化到 SQLite(重啟不遺失)
# scheduler = BackgroundScheduler(jobstores={'default': 'sqlite:///jobs.sqlite'})
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
scheduler.shutdown()
| 特性 | sched | threading.Timer | APScheduler |
|---|---|---|---|
| 依賴 | 標準庫 | 標準庫 | 第三方 |
| 阻塞性 | 阻塞 | 非阻塞 | 非阻塞(背景執行緒) |
| 重複排程 | 需自行重新 enter | 不支援 | 內建 interval/cron |
| 持久化 | 無 | 無 | 支援 (SQLite, Redis, MongoDB...) |
| 叢集/分散式 | 無 | 無 | 支援 (配合資料庫) |
| 管理介面 | 無 | 無 | 可整合 Flask/FastAPI 監控 |
| 適用場景 | 簡單腳本、一次性延遲 | 簡單非阻塞延遲 | 生產服務、定時任務、排程管理 |
threading.Timer 最輕量。sched 配合 while True: time.sleep(60) 輪詢。BackgroundScheduler 預設使用執行緒池(ThreadPoolExecutor),CPU 密集型任務請改用 ProcessPoolExecutor。misfire_grace_time 和 coalesce=True 避免疊加執行。logging)與監控(Prometheus + Grafana 或內建 scheduler.print_jobs())。掌握這三種工具的優劣,即可針對不同專案規模與需求,選出最合適的排程方案。