当前位置：首页>python>Python 初学者也能看懂的 APScheduler 入门指南

Python 初学者也能看懂的 APScheduler 入门指南

2026-06-29 14:00:07

PYTHON · SCHEDULER

Python 初学者也能看懂的 APScheduler 入门指南

从零理解定时任务：会写函数，就能学会 APScheduler

技术博客/ Python 入门 / 阅读 12 min

◎ 导语

当你写 Python 程序时，很快会遇到这样的问题：

✔ 每天凌晨 2 点清理一次日志。	✔ 每隔 10 分钟同步一次数据。
✔ 用户提交任务后，延迟 30 秒再执行。	✔ 后台长期运行一个定时巡检程序。

这些需求都属于“定时任务”。如果你只是简单地 while True 加 time.sleep()，刚开始能跑，但代码很快会变得不好维护。比如：怎么按每天 9 点执行？怎么暂停任务？怎么处理任务异常？程序重启后任务还在不在？

⚡ 先记住

这时就可以使用 APScheduler。

⚡ 先记住

APScheduler 的全称是 Advanced Python Scheduler，它是 Python 里非常常用的定时任务框架。它可以帮你用更清晰、更专业的方式管理定时任务。

⚡ 先记住

本文面向 Python 初学者，不要求你熟悉调度系统，只需要会写基本的 Python 函数。

01	/ APScheduler 能做什么

APScheduler 主要负责一件事：在指定的时间规则下，自动执行你写好的 Python 函数。

比如你有一个函数：

</> python

01def send_report():

02 print("发送日报")

你可以告诉 APScheduler：

每天早上 9 点执行它。
每隔 5 分钟执行它。
在某个具体时间点只执行一次。

APScheduler 会在后台帮你盯着时间，到点就调用这个函数。

02	/ 安装 APScheduler

使用 pip 安装即可：

</> bash

01pip install APScheduler

安装完成后，可以在 Python 里导入它：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

如果没有报错，说明安装成功。

最简单的例子

下面的例子会每隔 3 秒打印一次当前时间：

</> python

01from datetime import datetime

02from apscheduler.schedulers.blocking import BlockingScheduler

05def print_time():

06 print("当前时间：", datetime.now())

09scheduler = BlockingScheduler()

11scheduler.add_job(print_time, "interval", seconds=3)

13print("定时任务已启动")

14scheduler.start()

运行后，你会看到类似输出：

</> text

01定时任务已启动

02当前时间： 2026-06-23 10:00:03

03当前时间： 2026-06-23 10:00:06

04当前时间： 2026-06-23 10:00:09

这段代码里有几个重点：

✔ `print_time` 是你要定时执行的函数。	✔ `BlockingScheduler` 是调度器。
✔ `add_job()` 用来添加任务。	✔ `"interval"` 表示按固定间隔执行。
✔ `seconds=3` 表示每 3 秒执行一次。	✔ `scheduler.start()` 表示启动调度器。

04	/ APScheduler 的几个核心概念

刚开始学 APScheduler，最容易被各种名词绕晕。其实先记住下面几个就够了。

概念	中文理解	作用
Scheduler	调度器	负责管理和运行所有任务
Job	任务	你要定时执行的函数
Trigger	触发器	决定任务什么时候执行
Executor	执行器	决定任务如何被执行，比如线程池
JobStore	任务存储	决定任务保存在哪里，比如内存或数据库

初学阶段最常接触的是前三个：Scheduler、Job、Trigger。

三种常用触发器

APScheduler 里最常用的触发器有三种：

date
：指定某个时间点执行一次。
interval
：按固定时间间隔重复执行。
cron
：按类似 Linux crontab 的规则执行。

date：只执行一次

date 适合“在某个具体时间执行一次”的场景。

</> python

01from datetime import datetime, timedelta

02from apscheduler.schedulers.blocking import BlockingScheduler

05def once_job():

06 print("这个任务只执行一次")

09scheduler = BlockingScheduler()

11run_time = datetime.now() + timedelta(seconds=10)

12scheduler.add_job(once_job, "date", run_date=run_time)

14print("任务将在 10 秒后执行")

15scheduler.start()

这个任务会在程序启动 10 秒后执行一次，执行完就不会再重复执行。

interval：每隔一段时间执行

interval 适合“每隔多久执行一次”的场景。

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

04def sync_data():

05 print("同步数据中...")

08scheduler = BlockingScheduler()

10scheduler.add_job(sync_data, "interval", minutes=5)

12scheduler.start()

这个任务会每隔 5 分钟执行一次。

interval 常用参数包括：

参数	含义
`weeks`	间隔多少周
`days`	间隔多少天
`hours`	间隔多少小时
`minutes`	间隔多少分钟
`seconds`	间隔多少秒

cron：按日历规则执行

cron 适合“每天几点”“每周几”“每月几号”这种更像日历的规则。

例如：每天早上 9 点执行：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

04def morning_job():

05 print("早上 9 点任务执行")

08scheduler = BlockingScheduler()

10scheduler.add_job(morning_job, "cron", hour=9, minute=0)

12scheduler.start()

再比如：每周一到周五的 18:30 执行：

</> python

01scheduler.add_job(

02 morning_job,

03 "cron",

04 day_of_week="mon-fri",

05 hour=18,

06 minute=30,

07)

cron 常用参数包括：

参数	含义	示例
`year`	年	`2026`
`month`	月	`1-12`
`day`	日	`1-31`
`day_of_week`	星期	`mon-fri`
`hour`	小时	`0-23`
`minute`	分钟	`0-59`
`second`	秒	`0-59`

06	/ BlockingScheduler 和 BackgroundScheduler 的区别

初学者常看到两种调度器：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

02from apscheduler.schedulers.background import BackgroundScheduler

它们的区别很重要。

BlockingScheduler

BlockingScheduler 会阻塞当前程序。

也就是说，执行到：

</> python

01scheduler.start()

后面的代码就不会继续往下走了，程序会一直停在那里运行定时任务。

它适合单独写一个定时任务脚本：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

04def job():

05 print("任务执行")

08scheduler = BlockingScheduler()

09scheduler.add_job(job, "interval", seconds=5)

10scheduler.start()

BackgroundScheduler

BackgroundScheduler 会在后台线程中运行，不会阻塞主程序。

它适合和 Web 服务、桌面程序或其他长期运行的程序一起使用。

</> python

01import time

02from apscheduler.schedulers.background import BackgroundScheduler

05def job():

06 print("后台任务执行")

09scheduler = BackgroundScheduler()

10scheduler.add_job(job, "interval", seconds=5)

11scheduler.start()

13print("主程序继续运行")

15while True:

16 time.sleep(1)

如果你写的是一个单独的定时任务脚本，优先使用 BlockingScheduler。

如果你要把定时任务放进已有程序里，通常使用 BackgroundScheduler。

07	/ 给任务函数传参数

如果你的函数需要参数，可以通过 args 或 kwargs 传入。

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

04def send_message(username, content):

05 print(f"给 {username} 发送消息：{content}")

08scheduler = BlockingScheduler()

10scheduler.add_job(

11 send_message,

12 "interval",

13 seconds=10,

14 args=["小明", "记得提交日报"],

15)

17scheduler.start()

也可以使用 kwargs：

</> python

01scheduler.add_job(

02 send_message,

03 "interval",

04 seconds=10,

05 kwargs={

06 "username": "小明",

07 "content": "记得提交日报",

08 },

09)

08	/ 给任务设置 id

建议给重要任务设置 id，这样以后可以方便地查询、删除、暂停或恢复任务。

</> python

01scheduler.add_job(

02 send_message,

03 "interval",

04 seconds=10,

05 id="send_message_job",

06 args=["小明", "记得提交日报"],

07)

如果你重复添加同一个 id 的任务，APScheduler 默认会报错。开发时常用 replace_existing=True，表示如果任务已存在，就替换它：

</> python

01scheduler.add_job(

02 send_message,

03 "interval",

04 seconds=10,

05 id="send_message_job",

06 args=["小明", "记得提交日报"],

07 replace_existing=True,

08)

09	/ 暂停、恢复和删除任务

有了任务 id 后，就可以管理任务。

暂停任务

</> python

01scheduler.pause_job("send_message_job")

恢复任务

</> python

01scheduler.resume_job("send_message_job")

删除任务

</> python

01scheduler.remove_job("send_message_job")

查看所有任务

</> python

01jobs = scheduler.get_jobs()

03for job in jobs:

04 print(job.id, job.next_run_time)

10	/ 任务异常怎么处理

如果任务函数内部报错，APScheduler 不会直接让整个调度器崩溃，但你应该自己处理异常，方便排查问题。

</> python

01def sync_data():

02 try:

03 print("开始同步数据")

04 # 这里写真正的同步逻辑

05 result = 1 / 0

06 except Exception as e:

07 print("同步数据失败：", e)

在真实项目中，更推荐使用 logging 记录日志：

</> python

01import logging

03logging.basicConfig(level=logging.INFO)

04logger = logging.getLogger(__name__)

07def sync_data():

08 try:

09 logger.info("开始同步数据")

10 # 这里写真正的同步逻辑

11 except Exception:

12 logger.exception("同步数据失败")

logger.exception() 会自动记录异常堆栈，比单纯 print() 更适合排查问题。

11	/ 程序重启后任务还在吗

默认情况下，APScheduler 把任务保存在内存里。

这意味着：

程序运行时，任务存在。
程序停止后，任务消失。
程序重启后，需要重新添加任务。

如果你希望任务重启后仍然存在，就需要使用 JobStore，比如把任务保存到数据库。

常见方式是使用 SQLAlchemy：

</> bash

01pip install SQLAlchemy

示例：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

02from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

05def job():

06 print("持久化任务执行")

09jobstores = {

10 "default": SQLAlchemyJobStore(url="sqlite:///jobs.sqlite")

11}

13scheduler = BlockingScheduler(jobstores=jobstores)

15scheduler.add_job(

16 job,

17 "interval",

18 seconds=10,

19 id="persistent_job",

20 replace_existing=True,

21)

23scheduler.start()

这样任务信息会保存到 jobs.sqlite 文件中。

需要注意：使用持久化任务时，任务函数必须是可以被 Python 正常导入的函数。不要使用匿名函数 lambda，也不要使用临时定义的内部函数。

12	/ 常用配置：避免任务堆积

假设一个任务每 10 秒执行一次，但某次执行花了 30 秒，会发生什么？

如果不加限制，可能出现多个同类任务同时执行，甚至越积越多。APScheduler 提供了几个参数帮助你控制这种情况。

</> python

01scheduler.add_job(

02 sync_data,

03 "interval",

04 seconds=10,

05 id="sync_data_job",

06 max_instances=1,

07 coalesce=True,

08 misfire_grace_time=30,

09)

这几个参数的含义如下：

参数	作用
`max_instances=1`	同一个任务最多同时运行 1 个实例
`coalesce=True`	如果错过多次执行，只补执行一次
`misfire_grace_time=30`	如果任务错过执行时间不超过 30 秒，仍然允许执行

初学时可以先记住这套组合：如果任务不应该并发执行，就设置 max_instances=1。

时区问题

如果你的任务涉及“每天几点执行”，一定要注意时区。

可以在创建调度器时指定时区：

</> python

01from apscheduler.schedulers.blocking import BlockingScheduler

04scheduler = BlockingScheduler(timezone="Asia/Shanghai")

这样 cron 规则会按上海时区计算。

比如：

</> python

01scheduler.add_job(

02 job,

03 "cron",

04 hour=9,

05 minute=0,

06)

表示每天北京时间 9 点执行。

14	/ 在 Flask 中简单使用

如果你在 Flask 项目里想启动一个后台任务，可以使用 BackgroundScheduler。

</> python

01from flask import Flask

02from apscheduler.schedulers.background import BackgroundScheduler

05app = Flask(__name__)

08def check_status():

09 print("检查系统状态")

12scheduler = BackgroundScheduler(timezone="Asia/Shanghai")

13scheduler.add_job(

14 check_status,

15 "interval",

16 seconds=30,

17 id="check_status_job",

18 replace_existing=True,

19)

20scheduler.start()

23@app.route("/")

24def index():

25 return "Hello Flask"

28if __name__ == "__main__":

29 app.run(debug=False)

这里要注意：开发环境如果开启 Flask 的 debug=True，自动重载可能导致任务被启动两次。初学阶段可以先设置 debug=False，等理解清楚后再处理重载问题。

初学者常见问题

1. 为什么我的程序一运行就退出了

如果你使用的是 BackgroundScheduler，它不会阻塞主线程。

下面的代码可能会马上退出：

</> python

01from apscheduler.schedulers.background import BackgroundScheduler

04def job():

05 print("任务执行")

08scheduler = BackgroundScheduler()

09scheduler.add_job(job, "interval", seconds=5)

10scheduler.start()

因为主程序没有其他事情要做，执行完就结束了。

解决方式之一是让主程序保持运行：

</> python

01import time

03while True:

04 time.sleep(1)

如果你只是写独立脚本，更简单的方式是使用 BlockingScheduler。

2. 为什么任务执行了两次

常见原因是程序启动了两份，或者 Web 框架的开发模式开启了自动重载。

例如 Flask 的 debug=True，可能会导致代码被加载两次。解决时可以先关闭 debug，或者确保调度器只在主进程中启动。

3. 为什么数据库持久化任务报错

常见原因是任务函数不能被正常序列化。

使用持久化任务时，尽量这样写：

</> python

01def my_job():

02 print("任务执行")

避免这样写：

</> python

01scheduler.add_job(lambda: print("任务执行"), "interval", seconds=5)

也尽量不要把任务函数写成某个函数内部的局部函数。

4. APScheduler 适合所有后台任务吗

不适合。

APScheduler 适合单机程序里的定时任务。如果你需要分布式任务队列、任务重试、任务结果追踪、大量异步任务处理，通常要考虑 Celery、RQ、Arq 等任务队列工具。

简单理解：

工具	适合场景
APScheduler	定时执行 Python 函数
Celery	分布式异步任务、任务队列
schedule	更轻量的简单定时任务
crontab	操作系统级定时脚本

如果你只是想在 Python 程序里每隔一段时间执行某个函数，APScheduler 是很合适的选择。

一个完整示例

最后看一个稍微完整一点的例子：

</> python

01import logging

02from datetime import datetime

03from apscheduler.schedulers.blocking import BlockingScheduler

06logging.basicConfig(

07 level=logging.INFO,

08 format="%(asctime)s [%(levelname)s] %(message)s",

09)

10logger = logging.getLogger(__name__)

13def sync_data():

14 try:

15 logger.info("开始同步数据")

16 logger.info("当前时间：%s", datetime.now())

17 # 在这里编写真实的数据同步逻辑

18 logger.info("数据同步完成")

19 except Exception:

20 logger.exception("数据同步失败")

23def clean_logs():

24 logger.info("开始清理日志")

25 # 在这里编写真实的日志清理逻辑

26 logger.info("日志清理完成")

29def main():

30 scheduler = BlockingScheduler(timezone="Asia/Shanghai")

32 scheduler.add_job(

33 sync_data,

34 "interval",

35 minutes=10,

36 id="sync_data_job",

37 max_instances=1,

38 coalesce=True,

39 replace_existing=True,

40 )

42 scheduler.add_job(

43 clean_logs,

44 "cron",

45 hour=2,

46 minute=0,

47 id="clean_logs_job",

48 max_instances=1,

49 coalesce=True,

50 replace_existing=True,

51 )

53 logger.info("调度器启动")

54 scheduler.start()

57if __name__ == "__main__":

58 main()

这个例子里有两个任务：

sync_data
：每 10 分钟同步一次数据。
clean_logs
：每天凌晨 2 点清理日志。

同时它还设置了：

✔ `timezone="Asia/Shanghai"`：使用中国时区。	✔ `id`：方便管理任务。
✔ `max_instances=1`：避免同一个任务并发执行。	✔ `coalesce=True`：避免错过任务后大量补跑。
✔ `replace_existing=True`：避免重复添加同名任务时报错。

★ 总结	APScheduler 的核心并不复杂：写好函数，设定时间规则，让调度器负责按时调用。

APScheduler 是 Python 中非常实用的定时任务框架。对于初学者来说，可以先掌握下面这些内容：

✔ 用 `pip install APScheduler` 安装。	✔ 用 `BlockingScheduler` 写独立定时任务脚本。
✔ 用 `BackgroundScheduler` 把任务放到已有程序后台运行。	✔ 用 `date` 执行一次性任务。
✔ 用 `interval` 执行固定间隔任务。	✔ 用 `cron` 执行每天几点、每周几这类日历任务。
✔ 给任务设置 `id`，方便后续管理。	✔ 使用 `max_instances`、`coalesce` 避免任务堆积。
✔ 涉及每天几点执行时，记得设置时区。