Python 常用定時任務框架簡介

以下是對Python主流定時任務框架的詳細對比及使用指南，我們可結合場景需求可快速定位合適方案：

---

🧩 一、輕量級方案（適合簡單任務）

1. schedule庫

• 特點：語法直觀，無需復雜配置，適合小型腳本或低頻任務。

• 安裝

  uv add schedule
  

• 示例代碼：

  import schedule
  import timedef job():print(f"Task executed at {time.strftime('%Y-%m-%d %H:%M:%S')}")# 每10秒鐘執行一次
  schedule.every(10).seconds.do(job)
  # 每天16:00執行
  schedule.every().day.at("16:00").do(job)while True:schedule.run_pending()time.sleep(1)
  

  

• 適用場景：單機腳本、低頻任務（如每日報表生成）。

• 缺點：不支持持久化或動態任務管理。

---

?? 二、中級方案（平衡功能與復雜度）

2. APScheduler

• 核心優勢：

  • 支持三種觸發器：interval（固定間隔）、cron（類Linux Cron語法）、date（單次任務）。
  • 可持久化任務到數據庫（SQLite/MongoDB/Redis等）。
  • 提供異步調度器（AsyncIOScheduler）兼容asyncio。

• 安裝：

  uv add apscheduler
  

• 示例代碼：

  from apscheduler.schedulers.background import BackgroundScheduler
  import timedef job():print(f"Task executed at {time.strftime('%Y-%m-%d %H:%M:%S')}")# 創建調度器
  scheduler = BackgroundScheduler()
  # 每20秒執行（interval模式）
  scheduler.add_job(job, 'interval', seconds=20)
  # 每天16點08分執行（cron模式）
  scheduler.add_job(job, 'cron', hour=16, minute=8)scheduler.start()
  try:while True:time.sleep(2)
  except KeyboardInterrupt:scheduler.shutdown()
  

  

• 適用場景：需動態增刪任務（如Web后臺定時任務）、復雜調度邏輯（如每月最后一個工作日）。

3. Celery + Celery Beat

• 特點：分布式任務隊列，天然支持異步和橫向擴展，需搭配消息中間件（如Redis/RabbitMQ）。

• 安裝：

  uv add celery
  

  本文使用Redis故需要安裝上含有Redis訪問庫的版本，否則訪問Redis會報錯，使用如下語句：

  uv add celery[redis]
  

• 示例代碼：

    # tasks.pyfrom celery_app import app@app.taskdef add(x: int, y: int) -> int:result = x + yprint("add task result ", result)return result@app.taskdef multiply(x: int, y: int) -> int:result = x * yprint("multiply task ", result)return result# celery_app.pyfrom celery import Celeryfrom celery.schedules import crontabapp = Celery('celery_app', broker='redis://localhost:6379/0')# 設置時區為東八區app.conf.timezone = 'Asia/Shanghai'app.conf.update(result_backend='redis://localhost:6379/0',beat_schedule={'add-every-10-seconds': {'task': 'tasks.add','schedule': 10.0, # run every 10 seconds'args': (10, 10)},'multiply-at-1715': {'task': 'tasks.multiply','schedule': crontab(hour='17', minute='15'),'args': (4, 5)}},include=['tasks'])    
  

  該方案執行起來相對復雜了些：

  1. 首先啟動 Celery worker：celery -A celery_app worker --loglevel=info

  2. 【可選】如需觀察定時任務自動運行，可啟動 Celery beat：celery -A celery_app beat --loglevel=info

  3. 你也可以通過運行腳本手動觸發任務

  

  # main.py
  from tasks import add if __name__ == "__main__":result = add.delay(4, 5)print('Task result: ', result.get())
  

• 適用場景：分布式系統、高可用需求（如微服務集群定時任務）。

---

🚀 三、異步專用方案（高性能I/O密集型）

4. AsyncIOScheduler（APScheduler子類）

• 用途：在異步環境中調度協程任務。

• 示例：

  from apscheduler.schedulers.asyncio import AsyncIOScheduler
  import asyncio
  import time
  import osdef tick():print(f"Tick! The time is: {time.strftime('%Y-%m-%d %H:%M:%S')}")async def main():scheduler = AsyncIOScheduler()scheduler.add_job(tick, 'interval', seconds=5)scheduler.start()print("Press Ctrl+{} to exit".format("Break" if os.name == "nt" else "C"))while True:await asyncio.sleep(1000)if __name__ == "__main__":# Execution will block here until Ctrl+C (Ctrl+Break on Windows) is pressed.try:asyncio.run(main())except (KeyboardInterrupt, SystemExit):pass   
  

  

5. aiocron

• 特點：極簡異步Cron調度器，語法類似Unix cron。

• 安裝：

  uv add aiocron
  

• 示例：

  import asyncio
  import datetime
  import aiocron# 定義一個簡單的異步函數
  async def attime_job():"""這個協程會每分鐘被調用一次。"""print(f"[{datetime.datetime.now()}] 你好，世界！這是一個每分鐘執行的任務。")# 定義另一個需要參數的異步函數
  async def attime_job_with_args(name):"""這個協程會在每小時的第5分鐘被調用。"""print(f"[{datetime.datetime.now()}] 你好，{name}！這是一個帶參數的任務。")await asyncio.sleep(1) # 模擬一些 I/O 操作print(f"[{datetime.datetime.now()}] {name} 的任務執行完畢。")if __name__ == '__main__':print("程序啟動，等待定時任務執行... (按 Ctrl+C 退出)")# 1. 創建一個每分鐘執行的 cron 任務 ('* * * * * *')# cron 格式: a b c d e# a: 分鐘 (0-59)# b: 小時 (0-23)# c: 日 (1-31)# d: 月 (1-12)# e: 周幾 (0-6, 0是周日)aiocron.crontab('* * * * *', func=attime_job)# 2. 創建一個在每小時的第 5 分鐘執行的任務#    并向任務函數傳遞參數 'Python'aiocron.crontab('5 * * * *', func=attime_job_with_args, args=('Python',))# 創建并運行 asyncio 事件循環# loop.run_forever() 會一直運行，直到被停止loop = asyncio.get_event_loop()try:loop.run_forever()except KeyboardInterrupt:print("\n程序被用戶中斷。")finally:loop.close()print("事件循環已關閉。")
  

  

---

🔍 四、框架橫向對比

框架	適用場景	異步支持	持久化	分布式	學習成本
schedule	小型腳本、低頻任務	?	?	?	?
APScheduler	單機復雜調度、動態任務管理	?（選配）	?	?	??
Celery + Beat	分布式系統、高可用場景	?	?	?	???
AsyncIOScheduler	異步應用（如FastAPI/Starlette）	?	?	?	??
aiocron	極簡異步定時任務	?	?	?	?

💎 推薦策略：

• 快速驗證或簡單腳本 → schedule；
• 單機復雜調度（如動態啟停任務） → APScheduler；
• 分布式生產環境 → Celery；
• 異步應用（如FastAPI） → AsyncIOScheduler。

---

?? 五、避坑指南

1. 阻塞問題：
   • 避免在同步調度器（如BackgroundScheduler）中調用耗時同步任務，否則會阻塞主線程。異步任務需用AsyncIOScheduler。
2. 持久化配置：
   • APScheduler若需重啟后保留任務，務必配置jobstore（如SQLite）。
3. 時區陷阱：
   • 所有調度器默認使用UTC時間，需顯式設置時區（如scheduler = BackgroundScheduler(timezone="Asia/Shanghai")）。

---

💎 總結

根據場景復雜度與規模選擇框架：輕量級選schedule，單機靈活調度用APScheduler，分布式系統用Celery，異步生態選AsyncIOScheduler或aiocron。務必注意任務持久化和時區配置，避免生產環境故障。