python中怎么实现一个定时任务？-Python教程-PHP中文网

python中怎么实现一个定时任务？

穿越時空

发布： 2025-09-14 20:31:01

原创

877人浏览过

答案：选择定时任务方案需权衡需求复杂度与稳定性，APScheduler因支持持久化、多种调度方式及并发执行，适合生产环境。

python中怎么实现一个定时任务？

Python实现定时任务，方法其实不少，从最简单的循环加延时，到内置的

threading.Timer

登录后复制

、

sched

登录后复制

模块，再到功能强大的第三方库如

APScheduler

登录后复制

或分布式任务队列

Celery

登录后复制

，选择哪个主要看你的需求复杂度和对稳定性的要求。

解决方案

对于大多数实际应用场景，尤其是需要灵活调度和持久化的，

APScheduler

登录后复制

是我个人首选。它支持多种调度器（阻塞、非阻塞）、多种存储后端和执行器，非常灵活。

安装 APScheduler:

pip install apscheduler

登录后复制

基本用法（Cron 表达式与间隔调度）:

立即学习“Python免费学习笔记（深入）”；

from apscheduler.schedulers.background import BackgroundScheduler
import time
import datetime

def my_job(text):
    """一个简单的任务函数"""
    print(f"任务执行了！当前时间: {datetime.datetime.now()}, 参数: {text}")

# 初始化一个后台调度器，它会在一个单独的线程中运行
scheduler = BackgroundScheduler()

# 添加一个每5秒执行一次的任务
scheduler.add_job(my_job, 'interval', seconds=5, args=['Hello Interval Job'])

# 添加一个每天凌晨2点30分执行的任务 (使用Cron表达式)
# scheduler.add_job(my_job, 'cron', hour=2, minute=30, args=['Good Morning Cron Job'])

# 启动调度器
scheduler.start()
print('调度器已启动，按 Ctrl+C 退出...')

try:
    # 保持主线程运行，否则调度器会退出
    # 这里可以执行其他业务逻辑，或者只是简单地等待
    while True:
        time.sleep(2)
except (KeyboardInterrupt, SystemExit):
    scheduler.shutdown() # 优雅关闭调度器
    print('调度器已关闭。')

登录后复制

这里我用了

BackgroundScheduler

登录后复制

，它在一个独立的线程中运行，不会阻塞主程序。如果你需要一个阻塞式的调度器（比如你的整个程序就是个纯粹的定时任务服务），可以使用

BlockingScheduler

登录后复制

。

更简单的单次延迟任务：

threading.Timer

登录后复制

如果只是想在N秒后执行一个函数，

threading.Timer

登录后复制

其实更轻量、直接，适合一次性或简单的延迟场景。

import threading
import time

def delayed_task():
    print("这个任务在延迟后执行了！")

# 5秒后执行 delayed_task 函数
timer = threading.Timer(5, delayed_task)
timer.start()
print("定时器已启动，等待5秒...")
# timer.cancel() # 如果想取消任务，可以在任务执行前调用此方法

登录后复制

这个方法虽然简单，但它不提供持久化，也不支持复杂的调度模式，更适合一次性或简单的延迟场景。

在Python中，我们该如何选择最适合的定时任务方案？

这确实是个让人头疼的问题，因为选项太多了。在我看来，选择的核心在于你对“可靠性”、“灵活性”和“复杂度”的权衡。

对于简单到不能再简单的场景，比如就想等几秒钟再干点事，

time.sleep()

登录后复制

配合一个循环就够了。但这其实不是严格意义上的“定时任务”，更像是流程控制中的暂停。如果你想在后台异步执行一个函数，

threading.Timer

登录后复制

是个不错的选择，轻量、直接，但记住，它只执行一次，且不具备持久化能力。

稍微复杂一点，需要周期性执行，但又不想引入太多依赖，Python内置的

sched

登录后复制

模块可以考虑。它提供了一个事件调度器，你可以安排事件在未来的某个时间点执行。不过，它的API用起来稍微有点“学院派”，而且同样不提供持久化，程序一退出，所有预定的任务就都没了。我个人用得不多，觉得它在实际生产中不够“皮实”。

ViiTor实时翻译

AI实时多语言翻译专家！强大的语音识别、AR翻译功能。

116

查看详情

进入生产环境，或者说你对任务的调度有更精细的要求，

APScheduler

登录后复制

几乎是我的不二之选。它支持Cron风格、间隔（interval）和指定日期（date）三种调度方式，能满足绝大部分需求。更重要的是，它提供了多种

JobStore

登录后复制

（如内存、MongoDB、Redis、SQLAlchemy等），这意味着你的任务配置可以被持久化，即使程序崩溃重启，任务也能恢复。还有

Executors

登录后复制

的概念，你可以选择在线程池或进程池中执行任务，这对于避免任务阻塞主线程，或者处理CPU密集型任务非常有用。它的API设计也比较直观，学习成本不高。

当你的定时任务系统需要处理海量任务、分布式部署、高可用，并且对消息队列有依赖时，

Celery

登录后复制

这样的分布式任务队列就该登场了。它通常与消息代理（如RabbitMQ、Redis）配合使用，能够将任务分发到多个Worker上并行执行，提供了任务重试、结果存储、监控等一系列企业级功能。但相应的，它的部署和配置会比

APScheduler

登录后复制

复杂得多，引入的依赖也更多。如果你的系统还没有达到这个规模，直接上

Celery

登录后复制

可能会有点“杀鸡用牛刀”。

所以，我的建议是：从小处着手，如果

APScheduler

登录后复制

能满足，就用它。当你发现

APScheduler

登录后复制

的单点故障、扩展性瓶颈成为问题时，再考虑升级到

Celery

登录后复制

这样的分布式方案。

APScheduler在实际应用中如何处理任务的持久化和并发执行？

这是

APScheduler

登录后复制

之所以强大的关键点，也是它在生产环境能站稳脚跟的原因。

任务持久化 (Job Stores): 设想一下，你部署了一个定时任务服务，结果服务器突然重启了，或者你的Python程序崩溃了。如果任务没有持久化，那么所有已经计划好的未来任务都会丢失，这在生产环境中是绝对不能接受的。

APScheduler

登录后复制

通过

JobStore

登录后复制

机制解决了这个问题。它支持多种存储后端，比如：

```
MemoryJobStore
```
登录后复制
：这是默认的，任务只存在内存中，程序一关就没了，适合开发测试或不要求持久化的场景。
```
SQLAlchemyJobStore
```
登录后复制
：可以连接到各种关系型数据库（SQLite, PostgreSQL, MySQL等），将任务元数据存储在数据库中。这是生产环境中最常用的选择之一，因为它稳定且易于管理。
```
MongoDBJobStore
```
登录后复制
：如果你用MongoDB，这个就很方便。
```
RedisJobStore
```
登录后复制
：利用Redis进行存储，速度快。当你配置了非内存的
```
JobStore
```
登录后复制
后，
```
APScheduler
```
登录后复制
在启动时会从存储中加载所有未完成或未来计划的任务。这意味着即使程序重启，你的定时任务也能“记住”它们该做的事情。

示例 (使用 SQLAlchemyJobStore 和 SQLite):

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
import datetime
import time

def persistent_job():
    print(f"持久化任务执行了！时间: {datetime.datetime.now()}")

# 配置 Job Stores
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite') # 存储到当前目录下的jobs.sqlite文件
}
# 配置 Executors
executors = {
    'default': {'type': 'threadpool', 'max_workers': 20} # 使用线程池，最大20个工作线程
}
# 配置任务默认设置
job_defaults = {
    'coalesce': True, # 如果调度器错过了多次执行，只执行一次（聚合）
    'max_instances': 1 # 同一时间只允许此任务的一个实例运行
}

scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)

# 添加一个每10秒执行一次的任务，并给它一个ID，方便管理。
# replace_existing=True 在调试时很有用，可以确保每次启动时都更新或创建这个任务。
scheduler.add_job(persistent_job, 'interval', seconds=10, id='my_persistent_task', replace_existing=True)

scheduler.start()
print('持久化调度器已启动，任务已存储。')
try:
    while True:
        time.sleep(1)
except (KeyboardInterrupt, SystemExit):
    scheduler.shutdown()
    print('调度器已关闭。')

登录后复制

并发执行 (Executors): 定时任务的执行往往需要时间，如果一个任务执行时间过长，它可能会阻塞后续的任务执行，甚至阻塞调度器本身。

APScheduler

登录后复制

通过

Executor

登录后复制

来解决这个问题。

```
ThreadPoolExecutor
```
登录后复制
：这是默认的，任务会在一个线程池中执行。对于I/O密集型任务（如网络请求、文件读写），这是一个很好的选择，因为它不会受Python GIL的限制。
```
ProcessPoolExecutor
```
登录后复制
：任务会在一个进程池中执行。如果你有CPU密集型任务，或者任务可能会因为某些原因崩溃导致整个调度器受影响，使用进程池会更安全，因为每个任务都在独立的进程中运行，受GIL限制较小，且一个进程的崩溃通常不会影响其他进程。

通过配置