python定时任务方法总结
总结如下:
①用time的sleep方法
②用threading的Timer方法
③用sched模块
④用schedule
⑤用 Timeloop 库运行定时任务
⑥用apscheduler的BlockingScheduler方法
01 time模块
这是最低级的方法,任务还需要配合着while来使用。缺点就是无法执行不同时间间隔的任务。
import time
def task():
print("定时任务执行中...")
while True:
task()
time.sleep(60) # 休眠1min
02threading模块
threading能与方法配合使用,且能多线程执行任务,缺点是时间周期与函数执行的时间需要统计,没办法做到定时触发。
from threading import Timer
def func():
print('Hello world')
Timer(5,func).start()
func()
03使用sched模块
Python的sched模块提供了一个事件调度器,可以在指定的时间执行函数。这是一个更灵活的解决方案,可以安排多个任务
import time
import sched
s = sched.scheduler(time.time, time.sleep)
def task(sc):
print("定时任务执行中...")
s.enter(3600, 1, task, (s,))
s.run()
enter(delay, priority, action, argument),安排一个事件来延迟 delay 个时间单位。
cancel(event):从队列中删除事件。如果事件不是当前队列中的事件,则该方法将跑出一个 ValueError。
run():运行所有预定的事件。这个函数将等待(使用传递给构造函数的 delayfunc() 函数),然后执行事件,直到不再有预定的事件。
04schedule
schedule是一个Python库,它提供了一种非常简单的方式来执行定时任务。它的语法直观易懂,适用于各种定时任务需求。
import schedule
import time
def mtask():
print("定时任务执行中...")
schedule.every(1).hour.do(mtask)
while True:
schedule.run_pending()
time.sleep(1)
05用 Timeloop 库运行定时任务
是一个库,可用于运行多周期任务。这是一个简单的库,它使用 decorator 模式在线程中运行标记函数
import time
from timeloop import Timeloop
from datetime import timedelta
tl=Timeloop()
@tl.job(interval=timedelta(seconds=2))
def sample_job_every_2s():
print("2s job current time : {}".format(time.ctime()))
if __name__=="__main__":
tl.start(block=True)
06apscheduler框架
apscheduler比较全面,是强大的定时任务框架,可设置开始时间,结束时间,间隔时间。而且任务执行时间可以不计算,如果你的任务执行就需要耗费几秒,间隔执行虽然是对的,但是没办法做到每次的间隔都一样。apscheduler解决了这方面的烦恼。
from apscheduler.schedulers.blocking import BlockingScheduler
def func(x):
print(x)
sched = BlockingScheduler()
#间隔打印Hello world
sched.add_job(func, 'interval', start_date= '2023-04-02 00:00:00', seconds=300, args=['Hello', ])
sched.add_job(func, 'interval', start_date= '2023-04-02 00:00:01', seconds=300, args=['world', ])
0401使用Cron表达式进行任务调度
Cron表达式是一种灵活而强大的调度方式,可以实现各种复杂的任务调度需求。APScheduler提供了CronTrigger触发器,可以方便地使用Cron表达式进行任务调度。下面是一个使用Cron表达式的示例:
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
# 创建调度器对象
scheduler = BlockingScheduler()
# 定义任务函数
def task_func():
print("执行任务...")
# 创建Cron触发器
trigger = CronTrigger.from_crontab('0 0 12 * * ?')
#意:移除问号'?',因为APScheduler不支持Quartz格式的cron表达式
# 添加定时任务,函数对象传递,
scheduler.add_job(task_func, trigger)
# 启动调度器
scheduler.start()
在这个示例中,我们使用了CronTrigger触发器,并通过from_crontab方法传入Cron表达式。这里的Cron表达式是一个字符串,用于描述任务的执行时间规则。你可以根据Cron表达式的语法,灵活地定义任务的执行时间。
标准的cron表达式通常包含5个字段,分别代表:
分钟(0-59)
小时(0-23)
日期(1-31)
月份(1-12 或者 JAN-DEC)
星期(0-7 或者 SUN-SAT,其中0和7都代表周日)
0402任务调度与多线程
在实际应用中,有些任务可能会耗时较长,为了不阻塞调度器的正常运行,我们可以将这些耗时任务放在单独的线程中执行。下面是一个使用多线程执行任务的示例:
from apscheduler.schedulers.background import BackgroundScheduler
import threading
# 创建调度器对象
scheduler = BackgroundScheduler()
# 定义任务函数
def task_func():
print("执行任务...")
# 执行耗时任务
threading.Thread(target=long_running_task).start()
# 添加定时任务
scheduler.add_job(task_func, 'interval', seconds=10)
# 启动调度器
scheduler.start()
结语
本文介绍了使用Python进行任务调度和自动化的基本思路和实现细节。通过合理利用Python的任务调度和自动化工具,我们可以提高工作效率,减少重复性工作的负担。希望本文能够帮助你更好地利用Python进行任务调度和自动化,提高工作效率。