当前位置:   article > 正文

【超实用】用Python语言实现定时任务的八个方法,建议收藏!

【超实用】用Python语言实现定时任务的八个方法,建议收藏!

在日常工作中,我们常常会用到需要周期性执行的任务,一种方式是采用 Linux 系统自带的 crond 结合命令行实现。另外一种方式是直接使用Python。接下来整理的是常见的Python定时任务的八种实现方式。

图片

利用while True: + sleep()实现定时任务

位于 time 模块中的 sleep(secs) 函数,可以实现令当前执行的线程暂停 secs 秒后再继续执行。所谓暂停,即令当前线程进入阻塞状态,当达到 sleep() 函数规定的时间后,再由阻塞状态转为就绪状态,等待 CPU 调度。

图片

基于这样的特性我们可以通过while死循环+sleep()的方式实现简单的定时任务。

代码示例:

  1. import datetime
  2. import time
  3. def time_printer():
  4.     now = datetime.datetime.now()
  5.     ts = now.strftime('%Y-%m-%d %H:%M:%S')
  6.     print('do func time :', ts)
  7. def loop_monitor():
  8.     while True:
  9.         time_printer()
  10.         time.sleep(5)  # 暂停5秒
  11. if __name__ == "__main__":
  12.     loop_monitor()

主要缺点:

  • 只能设定间隔,不能指定具体的时间,比如每天早上8:00

  • sleep 是一个阻塞函数,也就是说 sleep 这一段时间,程序什么也不能操作。

使用Timeloop库运行定时任务

Timeloop是一个库,可用于运行多周期任务。这是一个简单的库,它使用decorator模式在线程中运行标记函数。

示例代码:

  1. import time
  2. from timeloop import Timeloop
  3. from datetime import timedelta
  4. tl = Timeloop()
  5. @tl.job(interval=timedelta(seconds=2))
  6. def sample_job_every_2s():
  7.     print "2s job current time : {}".format(time.ctime())
  8. @tl.job(interval=timedelta(seconds=5))
  9. def sample_job_every_5s():
  10.     print "5s job current time : {}".format(time.ctime())
  11. @tl.job(interval=timedelta(seconds=10))
  12. def sample_job_every_10s():
  13.     print "10s job current time : {}".format(time.ctime())

利用threading.Timer实现定时任务

threading 模块中的 Timer 是一个非阻塞函数,比 sleep 稍好一点,timer最基本理解就是定时器,我们可以启动多个定时任务,这些定时器任务是异步执行,所以不存在等待顺序执行问题。

Timer(interval, function, args=[ ], kwargs={ })

  • interval: 指定的时间

  • function: 要执行的方法

  • args/kwargs: 方法的参数

备注:Timer只能执行一次,这里需要循环调用,否则只能执行一次

案例:Python使用threading.Timer实现执行可循环的定时任务_python timer 循环-CSDN博客

利用内置模块sched实现定时任务

sched模块实现了一个通用事件调度器,在调度器类使用一个延迟函数等待特定的时间,执行任务。同时支持多线程应用程序,在每个任务执行后会立刻调用延时函数,以确保其他线程也能执行。

class sched.scheduler(timefunc, delayfunc)这个类定义了调度事件的通用接口,它需要外部传入两个参数,timefunc是一个没有参数的返回时间类型数字的函数(常用使用的如time模块里面的time),delayfunc应该是一个需要一个参数来调用、与timefunc的输出兼容、并且作用为延迟多个时间单位的函数(常用的如time模块的sleep)。

代码示例:

  1. import datetime
  2. import time
  3. import sched
  4. def time_printer():
  5.     now = datetime.datetime.now()
  6.     ts = now.strftime('%Y-%m-%d %H:%M:%S')
  7.     print('do func time :', ts)
  8.     loop_monitor()
  9. def loop_monitor():
  10.     s = sched.scheduler(time.time, time.sleep)  # 生成调度器
  11.     s.enter(51, time_printer, ())
  12.     s.run()
  13. if __name__ == "__main__":
  14.     loop_monitor()

scheduler对象主要方法:

  • enter(delay, priority, action, argument),安排一个事件来延迟delay个时间单位。

  • cancel(event):从队列中删除事件。如果事件不是当前队列中的事件,则该方法将跑出一个ValueError。

  • run():运行所有预定的事件。这个函数将等待(使用传递给构造函数的delayfunc()函数),然后执行事件,直到不再有预定的事件。

个人点评:比threading.Timer更好,不需要循环调用。

利用调度模块schedule实现定时任务

schedule是一个第三方轻量级的任务调度模块,可以按照秒,分,小时,日期或者自定义事件执行时间。schedule允许用户使用简单、人性化的语法以预定的时间间隔定期运行Python函数(或其它可调用函数)。

先来看代码,是不是不看文档就能明白什么意思?

  1. import schedule
  2. import time
  3. def job():
  4.     print("I'm working...")
  5. schedule.every(10).seconds.do(job)
  6. schedule.every(10).minutes.do(job)
  7. schedule.every().hour.do(job)
  8. schedule.every().day.at("10:30").do(job)
  9. schedule.every(5).to(10).minutes.do(job)
  10. schedule.every().monday.do(job)
  11. schedule.every().wednesday.at("13:15").do(job)
  12. schedule.every().minute.at(":17").do(job)
  13. while True:
  14.     schedule.run_pending()
  15.     time.sleep(1)

装饰器:通过 @repeat() 装饰静态方法

  1. import time
  2. from schedule import every, repeat, run_pending
  3. @repeat(every().second)
  4. def job():
  5.     print('working...')
  6. while True:
  7.     run_pending()
  8.     time.sleep(1)

传递参数:

  1. import schedule
  2. def greet(name):
  3.     print('Hello', name)
  4. schedule.every(2).seconds.do(greet, name='Alice')
  5. schedule.every(4).seconds.do(greet, name='Bob')
  6. while True:
  7.     schedule.run_pending()

装饰器同样能传递参数:

  1. from schedule import every, repeat, run_pending
  2. @repeat(every().second, 'World')
  3. @re
声明:本文内容由网友自发贡献,转载请注明出处:【wpsshop博客】
推荐阅读
相关标签
  

闽ICP备14008679号