首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式碼之家 >> 編程語言 >  >> 更多程式語言 >> 更多程式設計知識

Python 任務排程:sched、threading.Timer 與 APScheduler 實戰比較

     編輯:更多程式設計知識

在 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 在背景等待

優點:非阻塞、無額外依賴。
缺點:僅支援一次性任務、無重複排程、無管理介面、進程重啟失效。

方案三:APScheduler(生產級、支援持久化、分散式)

安裝: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()

功能對比表

特性schedthreading.TimerAPScheduler
依賴標準庫標準庫第三方
阻塞性阻塞非阻塞非阻塞(背景執行緒)
重複排程需自行重新 enter不支援內建 interval/cron
持久化無無支援 (SQLite, Redis, MongoDB...)
叢集/分散式無無支援 (配合資料庫)
管理介面無無可整合 Flask/FastAPI 監控
適用場景簡單腳本、一次性延遲簡單非阻塞延遲生產服務、定時任務、排程管理

選擇建議

  • 一次性腳本、測試用:threading.Timer 最輕量。
  • 單機長期運行服務、需 cron 風格、需持久化:直接上 APScheduler。
  • 分散式系統、多節點排程去重:APScheduler + Redis/MongoDB jobstore。
  • 極簡環境(無法安裝套件):sched 配合 while True: time.sleep(60) 輪詢。

進階提示

  • APScheduler 的 BackgroundScheduler 預設使用執行緒池(ThreadPoolExecutor),CPU 密集型任務請改用 ProcessPoolExecutor。
  • 若任務可能超過下一次觸發間隔,設定 misfire_grace_time 和 coalesce=True 避免疊加執行。
  • 生產環境務必加入日誌(logging)與監控(Prometheus + Grafana 或內建 scheduler.print_jobs())。

掌握這三種工具的優劣,即可針對不同專案規模與需求,選出最合適的排程方案。

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式碼之家 All Rights Reserved