Python实现定时任务的多种方案:BlockingScheduler、Timer、sched及系统级配置
本文详细解析Python中实现定时任务的多种方法,包括基础循环sleep、threading.Timer非阻塞定时器、sched调度模块以及功能强大的APScheduler框架。同时介绍如何在Windows和Linux环境下利用系统级工具如exe打包与Crontab配置定时任务,帮助开发者根据实际需求选择最合适的方案。
Python定时任务实现概述
本文介绍了Python中BlockingScheduler定时任务及其他多种实现方式,具体内容如下。
#BlockingScheduler定时任务
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
首先看看周一到周五定时执行任务
# 输出时间
def job():
print(datetime.now().strtime("%Y-%m-%d %H:%M:%S"))
# BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(job, "cron", day_of_week="1-5", hour=6, minute=30)
schduler.start()
scheduler.add_job(job, 'cron', hour=1, minute=5)
hour =19 , minute =23 这里表示每天的19:23 分执行任务
hour ='19', minute ='23' 这里可以填写数字,也可以填写字符串
hour ='19-21', minute= '23' 表示 19:23、 20:23、 21:23 各执行一次任务
#每300秒执行一次
scheduler .add_job(job, 'interval', seconds=300)
#在1月,3月,5月,7-9月,每天的下午2点,每一分钟执行一次任务
scheduler .add_job(func=job, trigger='cron', month='1,3,5,7-9', day='*', hour='14', minute='*')
# 当前任务会在 6、7、8、11、12 月的第三个周五的 0、1、2、3 点执行
scheduler .add_job(job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
#从开始时间到结束时间,每隔俩小时运行一次
scheduler .add_job(job, 'interval', hours=2, start_date='2018-01-10 09:30:00', end_date='2018-06-15 11:00:00')
#自制定时器
from datetime import datetime
import time
# 每n秒执行一次
def timer(n):
while True:
print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
time.sleep(n)
timer(5)
在Python中实现定时任务存在多种不同的技术路径。
基础实现:循环Sleep与Timer
第一种方法是循环配合sleep函数。
这是最基础的手段,即在循环体内放置需要执行的任务逻辑,随后调用sleep函数等待指定时间后再进行下一次执行。其主要缺点在于控制精度较低,且由于sleep是阻塞式函数,会占用当前线程资源。
def timer(n):
'''
每n秒执行一次
'''
while True:
print time.strftime('%Y-%m-%d %X',time.localtime())
yourTask() # 此处为要执行的任务
time.sleep(n)
第二种方法利用threading模块中的Timer类。
threading模块提供的Timer组件能够支持非阻塞式的定时任务执行。
例如,可以设置程序在3秒后打印“helloworld”。
def printHello():
print "hello world"
Timer(3, printHello).start()
或者设置为每隔3秒重复打印一次“helloworld”。
def printHello():
print "Hello World"
t = Timer(2, printHello)
t.start()
if __name__ == "__main__":
printHello()
系统级调度:sched模块
第三种方法是使用sched模块。
sched模块提供了一种基于事件的调度机制,用于处理延时任务。
# -*- coding:utf-8 -*-
# use sched to timing
import time
import os
import sched
# 初始化sched模块的scheduler类
# 第一个参数是一个可以返回时间戳的函数,第二个参数可以在定时未到达之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
# 被周期性调度触发的函数
def execute_command(cmd, inc):
'''
终端上显示当前计算机的连接情况
'''
os.system(cmd)
schedule.enter(inc, 0, execute_command, (cmd, inc))
def main(cmd, inc=60):
# enter四个参数分别为:间隔事件、优先级(用于同时间到达的两个事件同时执行时定序)、被调用触发的函数,
# 给该触发函数的参数(tuple形式)
schedule.enter(0, 0, execute_command, (cmd, inc))
schedule.run()
# 每60秒查看下网络连接情况
if __name__ == '__main__':
main("netstat -an", 60)
高级框架:APScheduler
第四种方法是采用定时框架APScheduler。
APScheduler是一个基于Quartz的Python定时任务框架,支持基于日期、固定时间间隔以及Crontab表达式类型的任务调度,并具备任务持久化能力。
作者目前尚未亲自尝试该框架,计划后续使用时再补充相关经验。
操作系统级定时任务配置
第五种方法是利用Windows系统的定时任务功能。
具体操作是将Python程序打包为exe可执行文件,随后在Windows系统中配置定时任务以执行该文件。
第六种方法是利用Linux系统的Crontab定时任务。
在Linux环境下,可以便捷地借助Crontab来配置和运行定时任务。通过进入Crontab文件编辑页面,设置执行时间间隔,并编写Shell命令来运行bash脚本或Python脚本,保存配置后Linux系统会自动按照设定时间定期执行程序。
以上即为本文的全部内容,希望能为读者的学习提供有益帮助,也欢迎大家继续支持本站。
免责声明:本文内容仅供参考,不代表本站观点。转载或引用请注明出处。









