本文转载自微信公众号「Python 作业辅导员」, 作者天元浪子. 转载本文请联系 Python 作业辅导员公众号.
所谓定时器, 是指间隔特定时间执行特定任务的机制. 几乎所有的编程语言, 都有定时器的实现. 比如, Java 有 util.Timer 和 util.TimerTask,JavaScript 有 setInterval 和 setTimeout, 可以实现非常复杂的定时任务处理. 然而, 牛叉到无所不能的 Python, 却没有一个像样的定时器, 实在令人难以理解.
刚入门的同学一定会说: 不是有个 time.sleep 吗? 定好闹钟睡大觉, 闹钟一响, 起来干活, 这不就是一个定时器吗? 没错, time.sleep 具备定时器的基本要素, 但若作为定时器使用, 则有两个致命的缺陷: 一是阻塞主线程, 睡觉的时候不能做任何事情; 二是醒来以后需要主线程执行定时任务 -- 即便使用线程技术, 也得先由主线程来创建子线程.
说到这里, 熟悉线程模块 threading 的同学也许会说: threading.Timer 就是以线程方式运行的呀, 既不会阻塞主线程, 执行定时任务也无需主线程干预, 这不就是一个完美的定时器吗?
我们先来看看 threading.Timer 是如何工作的. 下面这段代码演示了 threading.Timer 的基本用法: 启动定时器 2 秒钟后以线程方式调用函数 do_something, 在定时器等待的 2 秒钟内, 以及 do_something 运行期间, 主线程仍然可以做其他工作 -- 此处是从键盘读取输入, 借以阻塞主线程, 以便观察定时器的工作情况.
- import time
- import threading
- def do_something(name, gender='male'):
- print(time.time(), '定时时间到, 执行特定任务' )
- print('name:%s, gender:%s'%(name, gender))
- timer = threading.Timer(2, do_something, args=('Alice',), kwargs={'gender':'female'})
- timer.start()
- print(time.time(), '定时开始时间')
- input('按回车键结束 \ n') # 此处阻塞主进程
正如我们所期待的那样, 定时器启动 2 秒钟后, 函数 do_something 被调用, 这期间可以随时敲击回车键结束程序. 这段代码的运行结果如下.
1627438957.4297626 定时开始时间
按回车键结束
1627438959.4299397 定时时间到, 执行特定任务
name:Alice, gender:female
从使用效果看, threading.Timer 称得上是一款简洁易用的定时器. 不过, threading.Timer 存在明显的短板, 那就是不支持连续的定时任务, 比如, 每隔 2 秒钟调用一次 do_something 函数. 如果一定要用 threading.Timer 实现连续定时, 只能用类似嵌套的变通方法, 在 do_something 函数中再次启动定时器.
- import time
- import threading
- def do_something(name, gender='male'):
- global timer
- timer = threading.Timer(2, do_something, args=(name,), kwargs={'gender':gender})
- timer.start()
- print(time.time(), '定时时间到, 执行特定任务' )
- print('name:%s, gender:%s'%(name, gender))
- time.sleep(5)
- print(time.time(), '完成特定任务' )
- timer = threading.Timer(2, do_something, args=('Alice',), kwargs={'gender':'female'})
- timer.start()
- input('按回车键结束 \ n') # 此处阻塞主进程
这段代码重新定义了 do_something 函数, 在函数开始位置启动下一次的定时任务. 之所以放在开始位置, 是为了保证两次定时之间的时间间隔尽可能精确. 饶是如此, 下面的运行结果显示, 两次定时之间的时间间隔比设计的 2 秒钟多了大约 10 毫秒, 且误差是连续累计的, 重复执行 100 次, 误差将会超过 1 秒钟.
按回车键结束
1627440628.683803 定时时间到, 执行特定任务
name:Alice, gender:female
1627440630.6929214 定时时间到, 执行特定任务
name:Alice, gender:female
1627440632.707388 定时时间到, 执行特定任务
name:Alice, gender:female
1627440633.6890671 完成特定任务
1627440634.722474 定时时间到, 执行特定任务
name:Alice, gender:female
1627440635.7092102 完成特定任务
1627440636.7277966 定时时间到, 执行特定任务
name:Alice, gender:female
针对连续的定时任务, threading.Timer 的表现还算差强人意, 只是这种嵌套的写法完全颠覆了代码美学. 对于像我这样有代码洁癖的程序员来说, 是无法容忍和不可接受的. 在我看来, 一个完美的定时器应该满足以下 5 个条件, 具备下图所示的结构.
不阻塞主线程
同时支持单次定时和连续定时
以线程或进程方式执行定时任务
定时任务的线程或进程的创建, 运行, 不影响定时精度
足够精确的定时精度, 且误差不会累计
既然 Python 没有提供一个像样的定时器, 那就自己写一个吧. 下面这个定时器, 满足上面提到的 5 个条件, 最短时间间隔可以低至 10 毫秒, 且误差不会累计. 虽然还不够完美, 但无论结构还是精度, 都还说得过去.
- import time
- import threading
- class PyTimer:
- """定时器类"""
- def __init__(self, func, *args, **kwargs):
- """构造函数"""
- self.func = func
- self.args = args
- self.kwargs = kwargs
- self.running = False
- def _run_func(self):
- """运行定时事件函数"""
- th = threading.Thread(target=self.func, args=self.args, kwargs=self.kwargs)
- th.setDaemon(True)
- th.start()
- def _start(self, interval, once):
- """启动定时器的线程函数"""
- if interval < 0.010:
- interval = 0.010
- if interval < 0.050:
- dt = interval/10
- else:
- dt = 0.005
- if once:
- deadline = time.time() + interval
- while time.time() < deadline:
- time.sleep(dt)
- # 定时时间到, 调用定时事件函数
- self._run_func()
- else:
- self.running = True
- deadline = time.time() + interval
- while self.running:
- while time.time() < deadline:
- time.sleep(dt)
- # 更新下一次定时时间
- deadline += interval
- # 定时时间到, 调用定时事件函数
- if self.running:
- self._run_func()
- def start(self, interval, once=False):
- """ 启动定时器
- interval - 定时间隔, 浮点型, 以秒为单位, 最高精度 10 毫秒
- once - 是否仅启动一次, 默认是连续的
- """
- th = threading.Thread(target=self._start, args=(interval, once))
- th.setDaemon(True)
- th.start()
- def stop(self):
- """停止定时器"""
- self.running = False
定时器类 PyTimer 实例化时, 需要传入定时任务函数. 如果定时任务函数有参数, 也可以按照位置参数, 关键字参数的顺序一并提供. PyTimer 定时器提供 start 和 stop 两个方法, 用于启动和停止定时器. 其中 stop 方法不需要参数, start 则需要一个以秒为单位的定时间隔参数. start 还有一个布尔型的默认参数 once, 可以设置是否单次定时. once 参数的默认值为 False, 即默认连续定时; 如果需要单次定时, 只需要将 once 置为 true 即可.
- def do_something(name, gender='male'):
- print(time.time(), '定时时间到, 执行特定任务' )
- print('name:%s, gender:%s'%(name, gender))
- time.sleep(5)
- print(time.time(), '完成特定任务' )
- timer = PyTimer(do_something, 'Alice', gender='female')
- timer.start(0.5, once=False)
- input('按回车键结束 \ n') # 此处阻塞主进程
- timer.stop()
上面是使用 PyTimer 定时器以 0.5 秒钟的间隔连续调用函数 do_something 的例子. 这段代码的运行结果如下.
按回车键结束
1627450313.425347 定时时间到, 执行特定任务
name:Alice, gender:female
1627450313.9226055 定时时间到, 执行特定任务
name:Alice, gender:female
1627450314.421761 定时时间到, 执行特定任务
name:Alice, gender:female
1627450314.9243422 定时时间到, 执行特定任务
name:Alice, gender:female
1627450315.422722 定时时间到, 执行特定任务
name:Alice, gender:female
1627450315.9200313 定时时间到, 执行特定任务
name:Alice, gender:female
1627450316.4204514 定时时间到, 执行特定任务
name:Alice, gender:female
1627450316.9215539 定时时间到, 执行特定任务
name:Alice, gender:female
1627450317.4228196 定时时间到, 执行特定任务
name:Alice, gender:female
1627450317.9245899 定时时间到, 执行特定任务
name:Alice, gender:female
1627450318.42355 定时时间到, 执行特定任务
name:Alice, gender:female
1627450318.4393418 完成特定任务
1627450318.9251466 定时时间到, 执行特定任务
name:Alice, gender:female
1627450318.9395308 完成特定任务
1627450319.4242043 完成特定任务
1627450319.4242043 定时时间到, 执行特定任务
name:Alice, gender:female
1627450319.9253905 定时时间到, 执行特定任务
name:Alice, gender:female
1627450319.9411068 完成特定任务
1627450320.425871 完成特定任务
1627450320.425871 定时时间到, 执行特定任务
name:Alice, gender:female
虽然每个定时任务需要运行 5 秒钟, 但每隔 0.5 秒都会准时启动一个新的线程运行定时任务. 从记录可以看出, 尽管每次定时任务的启动时间有几个毫秒的误差, 但误差不会累计, 重复执行的时间间隔均值始终稳定在 0.5 秒.
来源: http://developer.51cto.com/art/202107/675442.htm