
你有没有遇到过这种场景?
数据每天凌晨3点要自动统计,缓存每隔10分钟要刷新,定时任务像闹钟一样在系统里此起彼伏。
说实话,很多Python开发者对定时任务的理解还停留在time.sleep层面。
其实Python提供了5种处理定时任务的方法,从最基础到最专业,从简单脚本到分布式系统,总有一款适合你。
今天这篇文章,就把这些方法一次性讲清楚。
往期阅读>>>
Python 自动化管理Jenkins的15个实用脚本,提升效率
App2Docker:如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率
如果你只是想快速实现一个简单的定时任务,schedule库是最佳选择。
它的语法非常直观,就像写自然语言一样:
schedule.every(10).minutes.do(job)
简单来说就是:每10分钟,执行一次job任务。
来看看具体怎么用:
import scheduleimport timedef job(): print("执行定时任务")# 设置调度规则schedule.every(10).minutes.do(job) # 每10分钟schedule.every().hour.do(job) # 每小时schedule.every().day.at("10:30").do(job) # 每天10:30while True: schedule.run_pending() time.sleep(1)这种方式的优点很明显:语法简单,上手快。
就像你手机上的闹钟应用,直观明了,不用看文档就能用。
如果你的任务特别简单,不想安装任何第三方库,threading.Timer就派上用场了。
这是Python标准库自带的,拿来就能用:
import threadingimport timedef 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()这种方式的特点是:零依赖,不需要安装任何东西。
就像你用手机自带的备忘录,虽然功能简单,但关键时刻总能帮你记下重要信息。
当你的定时任务变得复杂,schedule就显得力不从心了。
这时候你需要APScheduler,它支持更丰富的调度方式:
from apscheduler.schedulers.blocking import BlockingSchedulerdef 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:就像一个专业的日程管理软件,不仅能设置闹钟,还能管理复杂的时间表。
如果你的系统规模越来越大,单机运行已经不够用了。
这时候Celery就登场了,它是一个强大的分布式任务队列系统:
# 安装Celery和Redispip install celery redis# celery_app.pyfrom celery import Celeryfrom celery.schedules import crontabapp = Celery('tasks', broker='redis://localhost:6379/0')app.conf.beat_schedule = { 'every-30-seconds': { 'task': 'tasks.add', 'schedule': 30.0, 'args': (16, 16) }}# tasks.pyfrom celery_app import app@app.taskdef add(x, y): return x + y# 启动命令# celery -A celery_app worker --loglevel=info# celery -A celery_app beat --loglevel=infoCelery的强大之处在于支持分布式执行。
任务可以在多台服务器上并行执行,还能监控任务状态,处理任务失败的情况。
就像一个专业的物流配送系统,不仅知道每个包裹的当前位置,还能自动调配车辆。
最后说说最基础的方法,time.sleep加循环:
import timedef task(): print(f"任务执行时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")while True: task() # 休眠60秒 time.sleep(60)这种方式最简单,但功能也最弱。
精度差,功能有限,但胜在直观易懂,适合临时写个脚本测试。
就像你用手表掐时间,虽然不够精准,但偶尔用来倒个泡面时间还是够用的。
看了这么多方法,你可能有点晕了。
别急,给你一个简单的选择指南:
对于生产环境,我建议使用Celery或APScheduler这类功能更完善的库。
毕竟,定时任务往往承担着关键业务逻辑,稳定性和可靠性很重要。
定时任务看似简单,实则是分布式系统的基石之一。
从最基础的time.sleep到专业的分布式任务队列,每种方法都有其存在的价值。
就像工具箱里的扳手,虽然都能拧螺丝,但每个场景都需要最合适的那个。
选对工具,事半功倍。
想高效学习Python?下面三本精选好书满足你的不同需求!
《流畅的Python(第2版)》——Python进阶必读!深入讲解高级特性与最佳实践,适合想精进的开发者。
《Python从新手到高手》:初学者首选,系统学习全栈技能。
《Python数据分析:从零基础入门到案例实战》——数据科学利器!手把手教你用Python处理数据,实战案例学完就能用。
三本书均支持先用后付、运费险和7天无理由退货,放心购买!点击“购买”按钮,立即开启你的Python学习之旅吧!
https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c
