Python定时任务案例如何设置循环定时

wen python案例 2026-06-29 25

Python定时任务案例：如何设置循环定时任务？从基础到实战详解

目录导读

为什么需要循环定时任务？
Python实现定时任务的三大主流方案
案例1：使用time.sleep()实现简单循环
案例2：schedule库——最优雅的定时方案
案例3：APScheduler——企业级定时任务框架
常见问题与高频问答（FAQ）
性能优化与避坑指南

为什么需要循环定时任务？

在自动化脚本、数据采集、系统监控、邮件提醒等场景中，我们经常需要让程序每隔一段时间自动执行特定操作。

Python定时任务案例如何设置循环定时

每5分钟抓取一次股票价格
每天凌晨2点清理日志文件
每30秒检查一次服务器健康状态

Python提供了多种实现循环定时任务的方式，但不同方案在精度、资源占用、易用性上差异巨大，本文将通过3个完整案例,手把手教你实现循环定时任务。

Python实现定时任务的三大主流方案

方案	适用场景	优点	缺点
`time.sleep()`	简单循环，无需秒级精度	无依赖，代码极简	阻塞主线程，无法并发
`schedule`库	中等复杂度定时需求	语法优美，支持多种周期	依赖系统时间，精度一般
`APScheduler`	高并发、复杂调度逻辑	支持持久化、分布式、崩溃恢复	配置稍复杂，学习成本略高

SEO优化提示：本文所有案例代码均可直接复制运行，强调“开箱即用”。

案例1：使用`time.sleep()`实现简单循环

代码示例

import time
from datetime import datetime
def my_task():
    print(f"[{datetime.now()}] 任务执行中...")
while True:
    my_task()
    time.sleep(60)  # 每60秒执行一次

运行效果

[2025-04-01 10:00:00] 任务执行中...
[2025-04-01 10:01:00] 任务执行中...

局限与改进

阻塞问题：time.sleep()会让主线程暂停，无法执行其他代码。
解决方案：使用threading.Timer实现非阻塞定时：
```
import threading
```

def repeat_task(): my_task() threading.Timer(60, repeat_task).start()

repeat_task()


---
## 4. 案例2：`schedule`库——最优雅的定时方案
### 安装
```bash
pip install schedule

循环定时任务实现

import schedule
import time
from datetime import datetime
def job():
    print(f"定时任务触发: {datetime.now()}")
# 设置循环定时规则
schedule.every(1).minutes.do(job)       # 每分钟执行
schedule.every().hour.at(":30").do(job) # 每小时30分执行
schedule.every().day.at("10:30").do(job)# 每天10:30执行
while True:
    schedule.run_pending()
    time.sleep(1)  # 降低CPU占用

高级用法：设置循环次数

def job_with_count():
    if hasattr(job_with_count, 'count'):
        job_with_count.count += 1
    else:
        job_with_count.count = 1
    print(f"第{job_with_count.count}次执行")
    if job_with_count.count >= 5:
        return schedule.CancelJob  # 执行5次后取消
schedule.every(3).seconds.do(job_with_count)

为什么推荐？

语法直观：every().day.at("08:00") 像自然语言一样可读。
支持多种周期：秒、分、时、天、周、月灵活组合。
轻量级：单文件库,无外部依赖。

案例3：`APScheduler`——企业级定时任务框架

安装与导入

pip install apscheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
sched = BlockingScheduler()
@sched.scheduled_job('interval', seconds=10)
def timed_job():
    print(f"[{datetime.now()}] 每10秒执行")
@sched.scheduled_job('cron', hour=18, minute=30)
def cron_job():
    print(f"每天18:30执行")
sched.start()

三种触发器详解

触发器	说明	示例
`interval`	固定间隔循环	`seconds=60`
`cron`	类似Linux crontab	`hour=8, minute=0`
`date`	单次定时	`run_date='2025-05-01 10:00:00'`

实战：带异常重试的定时任务

from apscheduler.events import EVENT_JOB_ERROR
def retry_listener(event):
    if event.exception:
        print(f"任务失败，5秒后重试: {event.exception}")
        scheduler.reschedule_job(event.job_id, trigger='interval', seconds=5)
scheduler = BlockingScheduler()
scheduler.add_listener(retry_listener, EVENT_JOB_ERROR)

SEO优化要点：使用re关键字“循环定时任务异常处理”“APScheduler 重试机制”提升内容覆盖度。

常见问题与高频问答（FAQ）

Q1：循环定时任务如何停止？

schedule：return schedule.CancelJob 或外部设置标志位。
APScheduler：scheduler.remove_job(job_id) 或 scheduler.shutdown()。

Q2：定时任务如何传递参数？

# schedule
def greet(name):
    print(f"Hello, {name}")
schedule.every().minute.do(greet, name="Python")
# APScheduler
scheduler.add_job(func=greet, trigger='interval', seconds=60, args=("Python",))

Q3：多个定时任务如何并行执行？

schedule 默认串行，需配合 threading 或 multiprocessing 实现并行：
```
def job1(): print("任务1")
def job2(): print("任务2")
```

import threading threading.Thread(target=job1).start() threading.Thread(target=job2).start()

- **`APScheduler`**：默认使用线程池，支持最大并发数设置：  
```python
from apscheduler.executors.pool import ThreadPoolExecutor
executors = {'default': ThreadPoolExecutor(10)}

Q4：定时任务精度能到毫秒级吗？

time.sleep() 精度依赖操作系统，一般10-20ms。

APScheduler 配合 AsyncIO 可实现毫秒级：

from apscheduler.schedulers.asyncio import AsyncIOScheduler

性能优化与避坑指南

1 避免任务堆积

定时任务执行时间超过间隔时间时，后续任务会排队。
解决方案：使用 APScheduler 的 misfire_grace_time 参数控制允许的延迟时间。

2 日志与监控

import logging
logging.basicConfig(level=logging.INFO)
# APScheduler自带日志
scheduler.add_listener(lambda e: logging.info(f"任务事件: {e}"), EVENT_JOB_EXECUTED)

3 持久化与崩溃恢复

APScheduler 支持将任务存储在数据库（SQLite、PostgreSQL等）中，重启后自动恢复定时任务：

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
jobstores = {'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')}
scheduler = BlockingScheduler(jobstores=jobstores)

本文通过3个核心案例，从最基础的 time.sleep() 到企业级 APScheduler，完整覆盖了Python循环定时任务的实现方法,你应该根据场景选择：

快速原型 → schedule
高并发生产环境 → APScheduler
极简脚本 → time.sleep() + threading.Timer

掌握这些方法后，你可以轻松构建从数据采集到自动化运维的各种定时任务系统，如果本文对你有帮助,请收藏或分享给更多的开发者。

Python定时任务案例如何设置循环定时