43｜定时任务与 APScheduler

后台任务在请求触发时执行，定时任务则按预定时间表运行：每天凌晨清理日志、每小时同步一次监控数据、每 5 分钟检查服务健康状态。APScheduler 是 Python 最常用的定时任务库，支持多种触发器和任务存储后端。

一、安装与基础

uv add apscheduler

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

scheduler = BackgroundScheduler()

二、触发器类型

间隔触发（IntervalTrigger）

def check_health():
    print("检查服务健康状态...")

# 每 5 分钟执行一次
scheduler.add_job(
    check_health,
    trigger=IntervalTrigger(minutes=5),
    id="health_check",
    replace_existing=True,
)

参数	含义
seconds / minutes / hours / days / weeks	间隔单位

定时触发（CronTrigger）

def cleanup_logs():
    print("清理过期日志...")

# 每天凌晨 3 点执行
scheduler.add_job(
    cleanup_logs,
    trigger=CronTrigger(hour=3, minute=0),
    id="cleanup_logs",
)

Cron 表达式字段：

字段	范围	示例
year	四位年份	2026
month	1-12	1, 6-12
day	1-31	*/2（每两天）
week	1-53
day_of_week	0-6 或 mon-sun	mon-fri
hour	0-23	9, 14, 18
minute	0-59	0, 30
second	0-59	0

# 工作日早上 9 点和下午 6 点
CronTrigger(day_of_week="mon-fri", hour="9,18", minute=0)

# 每小时的第 0 和第 30 分钟
CronTrigger(minute="0,30")

日期触发（DateTrigger）

在指定时间执行一次：

from apscheduler.triggers.date import DateTrigger
from datetime import datetime

scheduler.add_job(
    send_reminder,
    trigger=DateTrigger(run_date=datetime(2026, 7, 1, 10, 0)),
)

三、与 FastAPI 集成

from fastapi import FastAPI
from contextlib import asynccontextmanager

scheduler = BackgroundScheduler()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：添加任务并启动调度器
    scheduler.add_job(check_health, IntervalTrigger(minutes=5), id="health")
    scheduler.start()
    yield
    # 关闭时：停止调度器
    scheduler.shutdown()

app = FastAPI(lifespan=lifespan)

lifespan 在应用启动时执行 yield 之前的代码，关闭时执行 yield 之后的代码。这是管理 APScheduler 生命周期的正确方式。

四、任务管理

移除任务

scheduler.remove_job("health_check")

暂停和恢复

scheduler.pause_job("health_check")
scheduler.resume_job("health_check")

获取任务列表

for job in scheduler.get_jobs():
    print(job.id, job.next_run_time)

五、任务持久化

默认任务只保存在内存中，进程重启后丢失。生产环境需要持久化：

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

jobstores = {
    "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")
}

scheduler = BackgroundScheduler(jobstores=jobstores)

任务信息（下次执行时间、执行次数等）存入数据库，进程重启后从数据库恢复。

六、并发执行

同一个任务如果上次还没执行完，下次触发时间又到了，默认会同时运行多个实例。如果任务不能并发，加锁：

import fcntl

def exclusive_task():
    with open("/tmp/task.lock", "w") as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
            # 执行任务...
        except BlockingIOError:
            print("上次任务还在执行，跳过")
            return
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)

七、常见错误

没有启动调度器

scheduler.add_job(check_health, IntervalTrigger(minutes=5))
# 错误：忘记 start()

# 正确
scheduler.start()

任务函数阻塞事件循环

# 错误：在异步应用中阻塞
async def task():
    time.sleep(60)   # 阻塞 60 秒，其他任务都卡住

# 正确：用异步 sleep
async def task():
    await asyncio.sleep(60)

时区问题

# 错误：没有指定时区，按系统本地时间执行
CronTrigger(hour=3)   # 服务器换时区后执行时间会变

# 正确：明确指定时区
from pytz import timezone
CronTrigger(hour=3, timezone=timezone("Asia/Shanghai"))