Python中实现定时任务

赵丹

Python中实现定时任务

在项目中，我们可能遇到有定时任务的需求。

• 其一：每隔一个时间段就执行任务。
  比如：压测中每隔45分钟调整温箱的温度。
  
• 其二：定时执行任务。
  例如每天早上 8 点定时推送早报。
  

今天，我跟大家分享下 Python 定时任务的实现方法。
固定时间间隔执行任务

1. import time
2. import logging
4. logging.basicConfig(
5.     level=logging.debug,
6.     format="%(asctime)s.%(msecs)d | %(threadName)s | %(levelname)s - %(message)s"
7. )
10. def task():
11.     logging.info("Task Start.")
12.     time.sleep(1)
13.     logging.info("Task Done.")

复制代码

time.sleep

第一种办法是最简单又最暴力。
那就是在一个死循环中，使用线程睡眠函数 sleep()。

1. while True:
2.     task()
3.     time.sleep(5)

复制代码

上述的方法有几个问题：

• 阻塞主进程，这个也好解决，可以放到线程中去执行
  
• 一次只有一个task，这个也有解决办法，可以多启几个线程
  

threading.Timer

既然第一种方法暴力，那么有没有比较优雅点的方法？
Python 标准库 threading 中有个 Timer 类。
它会新启动一个线程来执行定时任务，所以它是非阻塞函式。

原理：线程中预置一个finished的事件，通过finished.wait等待固定时间间隔。
超时则执行任务。如果需要取消任务，可以调用Timer.cancel来取消任务。

1. from threading import Timer
3. t = Timer(task, 5)
4. t.start()

复制代码

优点：
不阻塞主进程，task在线程中执行
缺点：
一个Timer只能执行一次task就结束了。
如果想循环，需要改造一下task函数。

1. from threading import Timer
3. def repeat_task():
4.     t = Timer(5, repeat_task)
5.     # 开始任务的位置决定了是任务之间等待固定间隔时间
6.     # 还是每个任务的开始等待固定间隔时间
7.     t.start()
8.     task()

复制代码

这样可以循环执行，但是仍然只能一个线程一个任务。

Q：如何跳出循环
A：加入一个标识符：

• sleep可以用一个布尔值来控制
  
• Timer可以用一个Event来控制
  

固定时间点执行任务

以上是用内置的方法中比较简单的实现方式。简单的功能可以实现。
前面执行的都是指定间隔时间的定时任务，那怎么执行指定时间点的任务呢？
上面的方法是可以做到的。有两种思路，

• 前面我们指定了间隔时间，那指定时间点，就先计算当前时间到指定时间点的间隔时间
  
• 不管间隔时间，而是记录任务时间点，然后实时去检查是否到达指定时间点
  

思路有了，但是对于固定时间点执行任务的场景以及后面更复杂的场景，自己实现可能就变得复杂。
下面再介绍几个进阶的定时任务的实现方式，可以适应更复杂的业务场景。
sched

第三种方式是使用标准库中sched模块。sched 是事件调度器，
它通过 scheduler 类来调度事件，从而达到定时执行任务的效果。
简单示例如下：

1. import sched
3. schedule = sched.scheduler()  # 初始化 sched 模块的 scheduler 类
4. schedule.enter(10, 1, task)  # 增加调度任务
5. schedule.run()  # 开始调度任务

复制代码

scheduler 提供了两个添加调度任务的函数:

• enter(delay, priority, action, argument=(), kwargs={})
  该函数可以延迟一定时间执行任务。delay 表示延迟多长时间执行任务，单位是秒。
  priority为优先级，越小优先级越大。两个任务指定相同的延迟时间，优先级大的任务会向被执行。
  action 即需要执行的函数，argument 和 kwargs 分别是函数的位置和关键字参数。
  
• scheduler.enterabs(time, priority, action, argument=(), kwargs={})
  添加一项任务，但这个任务会在 time 这时刻执行。因此，time 是绝对时间。其他参数用法与 enter() 中的参数用法是一致。
  

优点：

• 执行时间间隔和时间点执行任务
  
• 可以添加不同的任务
  
• 任务可以设置优先级
  

缺点：
scheduler 中的每个调度任务只会工作一次，不会无限循环被调用。如果想重复执行同一任务， 需要重复添加调度任务即可。

1. import sched
2. import threading
3. from functools import wraps
5. s = sched.scheduler()
6. STOP_FLAG = threading.Event()
8. def repeat(interval):
9.     def wrapper(func):
10.         @wraps(func)
11.         def inner():
12.             if not STOP_FLAG.is_set():
13.                 s.enter(interval, 0, inner)
14.             func()
15.         return inner
17.     return wrapper
20. @repeat(5)
21. def new_task():
22.     return task()
24. new_task()
25. t = threading.Thread(target=s.run)
26. t.start()

复制代码

实现原理

当然我们仅仅学会怎么用还是不够的，不能知其然而不知其所以然。
介绍了那么多方法，那么多库，但是底层的实现逻辑都是差不多的。
一个完整的定时任务系统，有三个部分：

• 任务队列(task queue)
  根据执行时间和优先级进行排序
  排序sorted(): 任务一多，整个队列重排，性能堪忧
  heapq.pop()：堆排序，只获取最高优先级的任务，不重复排序
  
• 调度器(scheduler)
  
• 执行器(executor)
  从前到后，实现的内容越来越多，控制的精细度也就越来越高，实现的功能也就越来越丰富。
  

1. import sched
2. import time
3. import threading
4. from functools import wraps
6. from task import task
7. from typing import Callable
9. STOP_FLAG = threading.Event()
12. def timeloop(func: Callable, interval: int):
13.     while True:
14.         func()
15.         time.sleep(interval)
18. def repeat_task(func: Callable, interval: int):
19.     # 新建任务的前后，决定了是在任务结束后等待时间，还是固定间隔
20.     if not STOP_FLAG.is_set():
21.         t = threading.Timer(interval, repeat_task, args=(func, interval))
22.         t.start()
23.     func()
26. def timer_schedule(func: Callable, interval: int):
27.     repeat_task(func, interval)
30. s = sched.scheduler()
33. def repeat(interval):
34.     def wrapper(func):
35.         @wraps(func)
36.         def inner():
37.             if not STOP_FLAG.is_set():
38.                 s.enter(interval, 0, inner)
39.             func()
40.         return inner
42.     return wrapper
45. @repeat(5)
46. def new_task():
47.     return task()
50. def sched_schedule(func: Callable, interval: int):
51.     func()
52.     t = threading.Thread(target=s.run)
53.     t.start()
54.     return t
57. def main(schedule):
58.     schedule_thread = schedule(new_task, 5)
59.     while True:
60.         try:
61.             time.sleep(1)
62.         except KeyboardInterrupt:
63.             STOP_FLAG.set()
64.             schedule_thread.join()
65.             break
68. if __name__ == "__main__":
69.     # main(timeloop)
70.     # main(timer_schedule)
71.     main(sched_schedule)

复制代码

来源:https://www.cnblogs.com/hd092336/p/17549357.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！