Web Day 21 排程任務:APScheduler 與 cron
執行需求:CPU 可跑;正式部署時需 Linux cron 或獨立的 scheduler 行程。今天是「Web 系統實戰:用 FastAPI 打造能上線的後端」系列的第二十一篇。昨天學了快取,把讀取路徑壓短了;今天換個方向:有些事不需要使用者按按鈕才做,例如「每天凌晨清掉過期的快取」、「每十分鐘把過期預約標為取消」、「每小時把統計資料彙整進資料庫」。這些都是排程任務。今天會用 APScheduler 3.11.x 把排程器掛進 FastAPI 的 lifespan,並對比 cron 在正式環境的差異。
引言
「排程任務」跟「背景任務」只有一字之差,但運作模式差很多。背景任務(Day 19)通常是被使用者動作觸發、馬上要排隊或立刻跑;排程任務則是「在固定的時間點」或「每隔一段時間」自動發生,跟使用者當下的行為無關。把這兩件事用同樣的工具做,雖然可行但會把程式碼搞混。今天介紹的 APScheduler 正是用來處理「與時間相關的事」。
今天的範例同樣接續 Day 16-20 的專案。我們會做四件事:第一,解釋 cron 表達式與 APScheduler 的對應;第二,用 lifespan 把 scheduler 掛進 FastAPI 應用;第三,寫兩個實際可跑的排程:每 30 秒清一次過期快取、每天凌晨發送當日預約提醒;第四,整理常見踩雷並討論 Windows 上的注意事項。
cron 表達式:五個欄位的時間規格
cron 是 Unix 系統幾十年來的標準排程格式,由五個(或六個含秒)欄位組成:
| 欄位 | 允許值 | 特殊語法 |
|---|---|---|
| 分鐘 | 0–59 | * , - / |
| 小時 | 0–23 | * , - / |
| 日 | 1–31 | * , - / ? |
| 月 | 1–12 或 jan–dec | * , - / |
| 週 | 0–6 或 sun–sat | * , - / ? |
幾個常見的範例:
*/5 * * * *:每 5 分鐘0 9 * * 1-5:週一到週五的早上 9 點30 2 * * *:每天凌晨 2:300 0 1 * *:每月 1 號午夜
APScheduler 內建 cron trigger,接受類似的表達式但加上了「秒」這個欄位(總共六個)。今天我們用 CronTrigger.from_crontab(...) 把標準五欄位字串直接吃進來,省去額外的學習成本。
APScheduler 與 FastAPI 的接法:lifespan
APScheduler 3.11.x 的核心物件是 BackgroundScheduler,它在背景啟動一個 thread 跑排程。我們用 FastAPI 的 lifespan 在應用啟動時啟動 scheduler、關閉時優雅停掉:
# app/scheduler.py
# 排程模組:把所有排程任務集中管理
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from app.cache import redis_client
def cleanup_expired_cache():
# 清掉所有 key 開頭是 web_day_20: 但沒有 TTL 的紀錄
# 真實情境可以用 SCAN + TTL 判斷,這裡用簡化版
keys = redis_client.keys("web_day_20:*")
if keys:
redis_client.delete(*keys)
print(f"[{datetime.now().isoformat()}] 清掉 {len(keys)} 個快取 key")
def daily_booking_digest():
# 每天寄一次預約摘要:真實情境會呼叫通知服務
print(f"[{datetime.now().isoformat()}] 寄出每日預約摘要")
scheduler = BackgroundScheduler(timezone="Asia/Taipei")
def configure_scheduler():
# 每 30 秒清一次過期快取(demo 用,正式環境可以每天一次)
scheduler.add_job(
cleanup_expired_cache,
trigger=IntervalTrigger(seconds=30),
id="cleanup_expired_cache",
replace_existing=True,
)
# 每天 23:00 寄當日預約摘要
scheduler.add_job(
daily_booking_digest,
trigger=CronTrigger.from_crontab("0 23 * * *"),
id="daily_booking_digest",
replace_existing=True,
)
scheduler.add_job 把函式與觸發器綁在一起,id 是任務的唯一識別碼,replace_existing=True 確保重複註冊時會覆蓋。FastAPI 的 lifespan 接上這個 scheduler:
# app/main.py
# 把 scheduler 接進 lifespan
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.scheduler import configure_scheduler, scheduler
configure_scheduler()
@asynccontextmanager
async def lifespan(app: FastAPI):
# 啟動:scheduler.start() 是非阻塞的
scheduler.start()
try:
yield
finally:
# 關閉:等所有正在跑的任務結束
scheduler.shutdown(wait=True)
app = FastAPI(title="Web Day 21 範例", lifespan=lifespan)
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/jobs")
def list_jobs():
# 把目前所有排程的 id 與下次執行時間列出來
jobs = []
for job in scheduler.get_jobs():
jobs.append({"id": job.id, "next_run": job.next_run_time.isoformat() if job.next_run_time else None})
return {"jobs": jobs}
lifespan 在 FastAPI 0.93 之後是推薦的寫法。啟動時呼叫 scheduler.start(),它會在背景 thread 開始跑排程;應用關閉時 scheduler.shutdown(wait=True) 會等所有任務跑完才結束,避免「任務跑到一半被砍掉」。/jobs 端點可以讓我們驗證排程真的有註冊、什麼時候會跑下一次。
完整實作:用 lifespan 把所有東西串起來
把前面幾天的工具(cache、notify、scheduler)組在一起:
# app/main.py(更完整的版本)
from contextlib import asynccontextmanager
from fastapi import BackgroundTasks, Depends, FastAPI
from pydantic import BaseModel
from app.cache import cache_aside, invalidate
from app.notify import Notifier, PrintNotifier, get_notifier
from app.scheduler import configure_scheduler, scheduler, daily_booking_digest
configure_scheduler()
@asynccontextmanager
async def lifespan(app: FastAPI):
scheduler.start()
try:
yield
finally:
scheduler.shutdown(wait=True)
app = FastAPI(title="Web Day 21 範例", version="0.21.0", lifespan=lifespan)
class Booking(BaseModel):
user_id: int
service: str
scheduled_at: str
@app.post("/bookings")
def create_booking(booking: Booking, background: BackgroundTasks, notifier: Notifier = Depends(get_notifier)):
# 真的預約邏輯(簡化版)
booking_id = hash((booking.user_id, booking.scheduled_at)) % 10_000
# 通知丟到背景
background.add_task(notifier.send, booking.user_id, f"預約 {booking.service} 成立")
return {"booking_id": booking_id}
@app.get("/jobs")
def list_jobs():
jobs = []
for job in scheduler.get_jobs():
jobs.append({"id": job.id, "next_run": job.next_run_time.isoformat() if job.next_run_time else None})
return {"jobs": jobs}
這個版本把 FastAPI lifespan、BackgroundTasks、Depends 全部串起來。啟動時 scheduler 開始跑、停掉時 scheduler 優雅關閉;建立預約時通知丟到背景;查 /jobs 端點可以驗證排程真的有註冊。
啟動的方式跟之前一樣:
# 命令列:啟動 FastAPI
uv run uvicorn app.main:app --reload --port 8000
# 預期輸出(節錄):
# [2025-08-05 12:00:00] 清掉 0 個快取 key
# INFO: Application startup complete.
啟動時你會看到第一個任務(cleanup_expired_cache)馬上跑了一次(APScheduler 的 IntervalTrigger 預設會立刻執行一次)。/jobs 端點回傳類似這樣:
# 命令列:列出目前排程
curl -s http://127.0.0.1:8000/jobs
# 輸出:{"jobs":[{"id":"cleanup_expired_cache","next_run":"2025-08-05T12:00:30+08:00"},{"id":"daily_booking_digest","next_run":"2025-08-05T23:00:00+08:00"}]}
測試排程任務:不要在測試裡真的等 30 秒
測試排程任務最常見的錯誤是「真的等 30 秒」。正解是「把觸發器換成幾秒後執行」或「手動呼叫任務函式」。下面這個範例展示後者:
# tests/test_scheduler.py
# 排程任務的測試:直接呼叫函式,而不是真的等
from datetime import datetime
from app.scheduler import cleanup_expired_cache, daily_booking_digest
def test_cleanup_expired_cache_removes_keys(fake_redis):
# 先塞一些資料
fake_redis.set("web_day_20:item:1", "x")
fake_redis.set("web_day_20:item:2", "y")
fake_redis.set("other:keep", "z")
assert fake_redis.exists("web_day_20:item:1") == 1
cleanup_expired_cache()
# web_day_20:* 都被清掉
assert fake_redis.exists("web_day_20:item:1") == 0
assert fake_redis.exists("web_day_20:item:2") == 0
# 其他 key 沒被影響
assert fake_redis.exists("other:keep") == 1
def test_daily_booking_digest_logs(monkeypatch, capsys):
logged = []
def fake_log(msg):
logged.append(msg)
monkeypatch.setattr("app.scheduler.print", fake_log)
daily_booking_digest()
assert any("每日預約摘要" in m for m in logged)
第一個測試直接呼叫 cleanup_expired_cache,不用等排程時間到;第二個測試用 monkeypatch 換掉 print,驗證函式有印出正確訊息。這種「測函式、不測排程」的寫法在 CI 環境特別穩定,幾百毫秒就能跑完。
如果你想測「排程真的有註冊」,可以另外寫一個測試:
# tests/test_scheduler_register.py
def test_scheduler_has_two_jobs():
from app.scheduler import scheduler
job_ids = {job.id for job in scheduler.get_jobs()}
assert "cleanup_expired_cache" in job_ids
assert "daily_booking_digest" in job_ids
這個測試只驗證「任務有被註冊」,不驗證「執行時間」。如果想驗證 cron 表達式有沒有寫錯,可以檢查 job.trigger 的內容,但這通常過於細節,留給 code review 處理就好。
Windows 上的注意事項
APScheduler 在 Windows 上運作得很好,跟 Linux 沒有明顯差異。但有兩個細節要記得:第一,BackgroundScheduler 預設用 thread 跑任務,Windows 對 thread 的管理與 Linux 略有差異,但對純計算或簡單 I/O 沒影響。第二,Windows 預設的 ProactorEventLoop 對 APScheduler 沒影響,因為排程本身是同步執行緒,與 asyncio 沒有交集。
如果你的任務會做長時間的同步 I/O(例如 requests.get),記得用 requests 而不是 httpx,因為 httpx 的 async client 在 Windows 上 ProactorEventLoop 的相容性還是有零星問題(Day 18 已提過)。對純計算的任務(例如「每天算一次彙總」),完全沒有作業系統差異。
另外一個常見問題:Windows 的時區。APScheduler 的 timezone 參數預設抓作業系統時區,Windows 可能回傳「太平洋標準時間」之類的歷史時區名稱。我們的範例明確指定 timezone="Asia/Taipei",這在 Windows 與 Linux 都會正確運作。
正式環境的選擇:APScheduler、cron、或獨立行程
當你部署到正式環境時,要把 APScheduler 想清楚:它跑在 FastAPI 行程裡,FastAPI 重啟、crash、scale out 都會影響排程。三個常見的策略:
- 保留 APScheduler(單機場景):當你只有一台伺服器、把 FastAPI 跑成單一行程時,APScheduler 是最簡單的選擇。我們的範例就是這種。
- 把排程拆到獨立的 worker:當你 scale out FastAPI(多個行程 + load balancer)時,如果每個行程都跑同一個排程,「清快取」會被執行好幾次。解法是把排程拆到一個獨立 worker,或用 Redis lock 確保只有一個行程執行某個任務。
- 改用作業系統 cron:當你只在 Linux 上跑、且排程不複雜,可以把 APScheduler 的任務寫成獨立的 .py 腳本,用 crontab 排時間觸發。cron 比 APScheduler 簡單,但無法動態新增任務、不能從程式裡改時間。
實務上,Day 35-44 的預約管理系統會用「FastAPI 行程跑 APScheduler 處理即時任務、cron 處理每日批次」。兩者分工:即時任務(每 30 秒清一次過期快取)交給 APScheduler;批次任務(每天凌晨寄月度報表)交給 cron 跑獨立腳本。這是兼顧彈性與穩定的組合。
常見錯誤與踩雷
第一個雷是「在 lifespan 之外呼叫 scheduler.start()」。APScheduler 的設計是「全域物件、啟動一次」,如果你在每個 request handler 或每個測試都呼叫 start(),第二次會拋 RuntimeError: Scheduler is already running。解法是把 scheduler 宣告成模組層級的全域物件,並在 lifespan 或測試的 setup 階段啟動一次。
第二個是「任務丟出例外把 scheduler 弄停」。當任務函式內部拋例外,APScheduler 預設會記 log、繼續跑下一個任務,但有些版本會把整個 scheduler 弄停。解法是在任務函式外面包一層 try/except、把例外寫進日誌(Web Day 22 會深入):
# app/scheduler.py(安全的任務寫法)
import logging
logger = logging.getLogger(__name__)
def cleanup_expired_cache():
try:
keys = redis_client.keys("web_day_20:*")
if keys:
redis_client.delete(*keys)
except Exception:
logger.exception("cleanup_expired_cache 失敗")
# 不要重新 raise,否則 scheduler 會以為整個 job 失敗
第三個是「時區混淆」。當你寫 CronTrigger(hour=9) 而不指定時區,APScheduler 會用 scheduler 預設時區。我們的範例用 timezone="Asia/Taipei",但如果 deploy 到 UTC 的容器、scheduler 用預設時區,「早上 9 點」會變成「UTC 9 點」而不是「台灣 9 點」。解法是顯式指定時區,並在 CI 環境用 TZ=Asia/Taipei 環境變數驗證。
第四個是「scheduler.shutdown() 阻塞太久」。當你在測試裡關閉 scheduler、任務還在跑,shutdown(wait=True) 會等到任務結束。實務上把 wait=False 設上去,shutdown 立刻返回,背景任務會被強制中斷。正式環境用 wait=True,測試環境用 wait=False,這是個常見的權衡。
效能與實務提醒
第一個提醒是「任務越短越好」。APScheduler 是單 thread 跑任務(預設),任務跑超過幾秒會擋住下一個任務的執行。多個任務要等前一個結束。實務上我們會讓任務「做完就結束」、「重的工作丟到背景 worker(Day 19)」,這是品質區塊一貫的分工。
第二個是「任務參數不要在閉包裡共用」。當你寫 scheduler.add_job(lambda: do_thing(state), ...),state 是閉包共用的變數,多個任務並行時可能互相覆寫。解法是把狀態放進函式參數,或用 thread-local storage。
第三個是「不要把排程寫死」。我們的範例把排程寫在程式碼裡,這對 demo 沒問題,但實務上應該把時間規格放進設定檔(Web Day 29)。當營運說「改成每天早上 8 點」時,你只需要改設定、不需要改程式碼、也不用重新部署。
APScheduler 的進階主題:job store 與持久化
APScheduler 預設的 job store 是記憶體版本(MemoryJobStore),應用重啟就會失去所有排程。對小型服務這通常沒問題(每次啟動都會重新註冊),但當你希望「應用死掉、job 不消失、重啟後繼續執行」,可以改用 SQLAlchemyJobStore 或 RedisJobStore:
# app/scheduler.py(使用 RedisJobStore)
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.schedulers.background import BackgroundScheduler
jobstores = {
"default": RedisJobStore(
jobs_key="web_day_21:jobs",
run_times_key="web_day_21:run_times",
host="localhost",
port=6379,
db=3,
)
}
scheduler = BackgroundScheduler(timezone="Asia/Taipei", jobstores=jobstores)
RedisJobStore 把每個 job 的設定(id、trigger、func reference)寫進 Redis;應用重啟後 scheduler 啟動時會自動把 job 載回來。這個寫法在多行程部署時非常實用:每個行程啟動時都能看到完整的排程清單,避免「只有第一個行程註冊了 job、其他行程沒有」。
小結
今天把排程任務的工作流建立起來了。APScheduler 3.11.x 透過 FastAPI 的 lifespan 啟動與關閉,搭配 IntervalTrigger 與 CronTrigger 處理「每 N 秒」與「在特定時間點」。我們寫了兩個示範任務:每 30 秒清一次快取、每天 23:00 寄預約摘要。測試上用「直接呼叫任務函式」取代「真的等排程時間」,讓 CI 維持在毫秒級。Windows 上的時區與事件迴圈問題透過顯式設定解決。正式環境的部署策略(單機 APScheduler、獨立 worker、作業系統 cron)也做了對比。今天結束時你的專案裡應該有一個「會自己跑」的排程器,啟動後不再只回應使用者請求、還會主動做事。
今晚的練習:加上你自己的排程
最快的練習方式是「從你的專案裡挑一個重複的任務,加成 APScheduler」。例如:「每小時刪除 24 小時前建立的測試資料」、「每 5 分鐘 ping 一個 health endpoint 並寫進資料庫」、「每週一早上 9 點清掉上週的 log」。
具體步驟:在 app/scheduler.py 寫一個新函式;用 scheduler.add_job 加進 configure_scheduler 裡;用 uv run uvicorn app.main:app --reload 啟動;用 /jobs 端點驗證註冊成功;等時間到,看 log 確認真的有跑。整個過程 30 分鐘內可以完成,跑完之後你就有一個「會自己跑」的服務,而不是只會等使用者按按鈕的程式。
如果你的任務需要資料庫或外部 API,記得在任務函式裡用 try/except 包起來(前面踩雷段落提過)。明天我們會把這些「會自己跑的事」與「會回應使用者的事」一起寫進日誌,讓你可以事後追溯問題。
排程與觀察性:當任務沒跑時怎麼辦
排程任務最常見的故障模式是「任務到了時間卻沒跑」。這通常有幾個原因:第一,行程死掉了。APScheduler 在背景 thread 跑,行程 OOM kill 就不會有下個 tick。第二,時區設錯了。排程「早上 9 點」但你以為是台灣 9 點、其實是 UTC 9 點。第三,任務函式拋例外被 scheduler 預設行為吞掉,於是看起來「沒跑」、其實是「跑失敗」了。
解法是「任務監控」。我們可以用 Web Day 22 即將介紹的 logging 模組,在任務開始與結束時印 INFO 等級的訊息,包含任務 id、執行時間、耗時。一旦任務失敗,至少能在 log 裡看到錯誤。實務上還會把任務的執行紀錄寫進資料庫或 Redis,這樣 dashboard 可以顯示「今天跑了幾次、最後一次是什麼時候」。這些觀察性設計在 Day 34「健康檢查與監控」會再深入展開。
APScheduler 與 asyncio:可以混合但要注意
APScheduler 的 BackgroundScheduler 用的是 thread,不是 asyncio。當你在 async 端點裡呼叫 scheduler 的方法(例如 scheduler.add_job(...)),它會在呼叫端點的 worker thread 同步執行,不會打斷事件迴圈。但如果你想讓排程任務本身是 async 函式,APScheduler 3.11.x 並不直接支援;正確做法是「在 async 任務裡用 asyncio.run(...) 啟動一個小事件迴圈」、或乾脆「用 APScheduler 觸發背景任務、把 async 工作交給 asyncio.create_task」。
這個混合模型對小型服務來說太複雜,實務上把排程任務寫成普通 def 函式最簡單。Day 35-44 的預約管理系統裡,所有排程任務都是普通函式:它們呼叫 SQLModel 同步 engine 抓資料、把結果寫進 Redis 8 或資料庫。async 端點則由 FastAPI 自己處理,兩者互不干擾。
把 APScheduler 換成 Celery beat:另一條路
如果你的排程任務需要「跨機器執行」、「失敗重試」、「從管理介面查狀態」,APScheduler 不夠用,可以考慮 Celery beat(Day 19 提過的 Celery 5.5 家族成員)。Celery beat 是 Celery 的排程器,搭配 Redis 或 RabbitMQ 作為 broker,能把任務交給 worker 行程執行。設定成本比 APScheduler 高,但橫向擴展能力強很多。
判斷原則跟 Day 19 一樣:「只有單機、任務簡單、無須重試」用 APScheduler;「多機、需要可靠傳遞、需要狀態查詢」用 Celery beat。Web Day 41「專案:Docker Compose 一鍵上線」會在預約管理系統專案裡同時用上兩者:APScheduler 處理即時的小任務、Celery beat 處理跨機器的批次任務。
結語
今天的排程任務跟昨天的背景任務都是「服務在沒人按的時候也會做事」。明天,Web Day 22「日誌設計與請求追蹤」會把這些「事」記錄下來:任務什麼時候跑、跑了多久、發生了什麼錯誤、誰發起的請求。今天我們已經在任務裡用 print,印在終端機看起來 OK,但正式環境需要更結構化的記錄方式,這正是明天的主軸。
延伸資源
- APScheduler 官方文件(3.11.x,2025):
https://apscheduler.readthedocs.io/en/3.x/ - FastAPI lifespan 官方教學(0.116,2025-07):
https://fastapi.tiangolo.com/advanced/events/ - cron 表達式速查表(crontab.guru,2025):
https://crontab.guru/ - Linux cron 官方說明(2025):
https://man7.org/linux/man-pages/man5/crontab.5.html - APScheduler 在容器環境的注意事項(2025):
https://apscheduler.readthedocs.io/en/3.x/faq.html
留言
張貼留言