Python定时任务5种方法，从入门到精通

用户11081884

发布于 2026-07-20 19:58:06

230

文章被收录于专栏：用户11081884的专栏用户11081884的专栏

你有没有遇到过这种场景？

数据每天凌晨3点要自动统计，缓存每隔10分钟要刷新，定时任务像闹钟一样在系统里此起彼伏。

说实话，很多Python开发者对定时任务的理解还停留在time.sleep层面。

其实Python提供了5种处理定时任务的方法，从最基础到最专业，从简单脚本到分布式系统，总有一款适合你。

今天这篇文章，就把这些方法一次性讲清楚。

1. schedule - 新手最快上手

如果你只是想快速实现一个简单的定时任务，schedule库是最佳选择。

它的语法非常直观，就像写自然语言一样：

schedule.every(10).minutes.do(job)

简单来说就是：每10分钟，执行一次job任务。

来看看具体怎么用：

import schedule
import time

def job():
    print("执行定时任务")

# 设置调度规则
schedule.every(10).minutes.do(job)  # 每10分钟
schedule.every().hour.do(job)      # 每小时
schedule.every().day.at("10:30").do(job)  # 每天10:30

while True:
    schedule.run_pending()
    time.sleep(1)

这种方式的优点很明显：语法简单，上手快。

就像你手机上的闹钟应用，直观明了，不用看文档就能用。

2. threading.Timer - 零依赖方案

如果你的任务特别简单，不想安装任何第三方库，threading.Timer就派上用场了。

这是Python标准库自带的，拿来就能用：

import threading
import time

def repeated_task():
    print("周期任务执行")
    # 重新设置定时器实现循环执行
    threading.Timer(5.0, repeated_task).start()

def one_time_task():
    print("一次性任务执行")

# 一次性任务，5秒后执行
threading.Timer(5.0, one_time_task).start()

# 循环任务，立即开始，每5秒执行一次
repeated_task()

这种方式的特点是：零依赖，不需要安装任何东西。

就像你用手机自带的备忘录，虽然功能简单，但关键时刻总能帮你记下重要信息。

3. APScheduler - 专业级调度器

当你的定时任务变得复杂，schedule就显得力不从心了。

这时候你需要APScheduler，它支持更丰富的调度方式：

from apscheduler.schedulers.blocking import BlockingScheduler

def job_function():
    print("定时任务执行中...")

scheduler = BlockingScheduler()
# 每5秒执行一次
scheduler.add_job(job_function, 'interval', seconds=5)
# 每天10:30执行
scheduler.add_job(job_function, 'cron', hour=10, minute=30)

scheduler.start()

APScheduler的核心优势是功能丰富且灵活。

支持定时、循环、一次性任务，还能持久化任务状态，重启后任务不会丢失。

APScheduler：就像一个专业的日程管理软件，不仅能设置闹钟，还能管理复杂的时间表。

4. Celery - 分布式任务队列

如果你的系统规模越来越大，单机运行已经不够用了。

这时候Celery就登场了，它是一个强大的分布式任务队列系统：

# 安装Celery和Redis
pip install celery redis

# celery_app.py
from celery import Celery
from celery.schedules import crontab

app = Celery('tasks', broker='redis://localhost:6379/0')

app.conf.beat_schedule = {
    'every-30-seconds': {
        'task': 'tasks.add',
        'schedule': 30.0,
        'args': (16, 16)
    }
}

# tasks.py
from celery_app import app

@app.task
def add(x, y):
    return x + y

# 启动命令
# celery -A celery_app worker --loglevel=info
# celery -A celery_app beat --loglevel=info

Celery的强大之处在于支持分布式执行。

任务可以在多台服务器上并行执行，还能监控任务状态，处理任务失败的情况。

就像一个专业的物流配送系统，不仅知道每个包裹的当前位置，还能自动调配车辆。

5. time.sleep - 最基础的实现

最后说说最基础的方法，time.sleep加循环：

import time

def task():
    print(f"任务执行时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")

while True:
    task()
    # 休眠60秒
    time.sleep(60)

这种方式最简单，但功能也最弱。

精度差，功能有限，但胜在直观易懂，适合临时写个脚本测试。

就像你用手表掐时间，虽然不够精准，但偶尔用来倒个泡面时间还是够用的。

到底该选哪种？

看了这么多方法，你可能有点晕了。

别急，给你一个简单的选择指南：

方法	适用场景	优点	缺点
Celery	分布式系统、复杂任务	功能强大、支持分布式	配置复杂
APScheduler	中等复杂度应用	功能丰富、灵活	需要学习API
schedule	简单应用、快速开发	语法直观、易用	功能相对简单
threading.Timer	简单单次/循环任务	无需额外依赖	功能有限
time.sleep	最简单的定时需求	零依赖、易理解	精度差、功能弱