Python 实现定时任务的八种方案 _Python

作者：钱魏Way
来源：https://www.biaodianfu.com/Python-schedule.html
在日常工作中，我们常常会用到需要周期性执行的任务，一种方式是采用 linux 系统自带的 crond 结合命令行实现。另外一种方式是直接使用Python 。接下里整理的是常见的Python定时任务的实现方式。

文章插图

目录

利用while True: + sleep()实现定时任务
使用Timeloop库运行定时任务
利用threading.Timer实现定时任务
利用内置模块sched实现定时任务
利用调度模块schedule实现定时任务
利用任务框架APScheduler实现定时任务 APScheduler中的重要概念 Job 作业Trigger 触发器Executor 执行器Jobstore 作业存储Event 事件调度器 Scheduler的工作流程
使用分布式消息系统Celery实现定时任务
使用数据流工具Apache Airflow实现定时任务 Airflow 产生的背景Airflow 核心概念Airflow 的架构

利用while True: + sleep()实现定时任务位于 time 模块中的 sleep(secs) 函数，可以实现令当前执行的线程暂停 secs 秒后再继续执行。所谓暂停，即令当前线程进入阻塞状态，当达到 sleep() 函数规定的时间后，再由阻塞状态转为就绪状态，等待 CPU 调度。
基于这样的特性我们可以通过while死循环+sleep()的方式实现简单的定时任务。
代码示例：

import datetimeimport timedef time_printer():now = datetime.datetime.now()ts = now.strftime('%Y-%m-%d %H:%M:%S')print('do func time :', ts)def loop_monitor():while True:time_printer()time.sleep(5)# 暂停5秒if __name__ == "__main__":loop_monitor()

主要缺点：

只能设定间隔，不能指定具体的时间，比如每天早上8:00
sleep 是一个阻塞函数，也就是说 sleep 这一段时间，程序什么也不能操作。

使用Timeloop库运行定时任务Timeloop是一个库，可用于运行多周期任务。这是一个简单的库，它使用decorator模式在线程中运行标记函数。
示例代码：
【Python 实现定时任务的八种方案】

import timefrom timeloop import Timeloopfrom datetime import timedeltatl = Timeloop()@tl.job(interval=timedelta(seconds=2))def sample_job_every_2s():print "2s job current time : {}".format(time.ctime())@tl.job(interval=timedelta(seconds=5))def sample_job_every_5s():print "5s job current time : {}".format(time.ctime())@tl.job(interval=timedelta(seconds=10))def sample_job_every_10s():print "10s job current time : {}".format(time.ctime())

利用threading.Timer实现定时任务threading 模块中的 Timer 是一个非阻塞函数，比 sleep 稍好一点，timer最基本理解就是定时器，我们可以启动多个定时任务，这些定时器任务是异步执行，所以不存在等待顺序执行问题。
Timer(interval, function, args=[ ], kwargs={ })

interval: 指定的时间
function: 要执行的方法
args/kwargs: 方法的参数

代码示例：

import datetimefrom threading import Timerdef time_printer():now = datetime.datetime.now()ts = now.strftime('%Y-%m-%d %H:%M:%S')print('do func time :', ts)loop_monitor()def loop_monitor():t = Timer(5, time_printer)t.start()if __name__ == "__main__":loop_monitor()

备注：Timer只能执行一次，这里需要循环调用，否则只能执行一次
利用内置模块sched实现定时任务sched模块实现了一个通用事件调度器，在调度器类使用一个延迟函数等待特定的时间，执行任务。同时支持多线程应用程序，在每个任务执行后会立刻调用延时函数，以确保其他线程也能执行。
class sched.scheduler(timefunc, delayfunc)这个类定义了调度事件的通用接口，它需要外部传入两个参数，timefunc是一个没有参数的返回时间类型数字的函数(常用使用的如time模块里面的time)，delayfunc应该是一个需要一个参数来调用、与timefunc的输出兼容、并且作用为延迟多个时间单位的函数(常用的如time模块的sleep) 。
代码示例：

import datetimeimport timeimport scheddef time_printer():now = datetime.datetime.now()ts = now.strftime('%Y-%m-%d %H:%M:%S')print('do func time :', ts)loop_monitor()def loop_monitor():s = sched.scheduler(time.time, time.sleep)# 生成调度器s.enter(5, 1, time_printer, ())s.run()if __name__ == "__main__":loop_monitor()

scheduler对象主要方法: