引言

Python的threading模块中，相关的同步原语我们都已经有所涉及，感兴趣的同学可以翻看前面的几篇文章。今天的这篇文章中，我们来介绍一个相对简单，但又比较实用的组件Timer。

本文的主要内容有：

1、什么是Timer

2、基于Timer实现任务的延迟执行

3、基于Timer实现周期性任务的调度执行

什么是Timer

Timer是Python的threading模块中提供的一个定时器的组件，可以实现在指定的时间间隔后执行一个函数。

从源码看，Timer模块的实现非常简单。

从源码中可以得到如下信息：

1、Timer的实现，主要是通过继承Thread类型，实现相关线程实例化、调度的基本功能。

2、内部持有一个Event事件对象，通过对象的指定timeout的wait()方法调用，实现等待指定时间再执行的效果。

3、提供cancel()方法，用于设置内部事件对象的标志位，从而终止等待执行。

虽然TImer的代码行数比较少，且使用起来比较简单，但是，其实还是有很多实用的使用场景的：

1、任务的延时执行：比如用户注册完成后，延迟几分钟后，异步发送注册成功的欢迎邮件；某些任务执行产生的临时文件，在任务执行完成后，定时进行清理等。

2、游戏中的定时器：比如在游戏中，可以设置定时器进行积分的累加，或者定时生成敌人等。

3、实现自动重试机制：比如任务失败后，等待3秒进行自动重试的场景。

4、间隔性轮询：通过定时器实现在后台执行定期的状态检查等。

总的来说，Timer的延迟执行可以用于各种业务场景，特别是在需要控制任务执行时间或者频率时非常方便。通过合理利用Timer，可以提高程序的响应性和用户体验，减少资源的浪费，并实现更急灵活、便捷的任务调度逻辑。

基于Timer实现任务的延迟执行

首先通过代码简单演示Timer的延时执行的功能，直接看代码：

from threading import Timer


def send_verify_code(code):
    print(f'验证码是：{code}')


if __name__ == '__main__':
    print('用户注册最后一步，请输入验证码')
    timer = Timer(interval=2, function=send_verify_code, args=('345231',))
    timer.start()

执行结果：

当然，代码中只是做了一个简单的场景模拟。在真实业务场景中，其实也是需要把相关的任务进行分离，核心、实时性要求高的任务同步进行处理，非实时性的任务通过消息队列等中间件实现异步、延迟执行，从而提高服务的高并发需求。

基于Timer实现周期性任务的调度执行

虽然Timer在实现上似乎只提供了延迟执行的特性，但是，我们可以通过延迟执行的函数体内继续连锁性的创建并启动新的Timer实例，从而可以实现周期性调度执行的功能。

首先，通过一个时间显示更新的代码来演示周期性调度执行的功能实现：

from threading import Timer
import datetime
import sys


def clock():
    # flush()这行可以没有，只是尽量保证输出能够及时从输出缓冲区中刷新到显示器
    sys.stdout.flush()
    print(f'\r当前时间为：{datetime.datetime.now().strftime("%X")}', end='')
    timer = Timer(interval=1, function=clock)
    timer.start()


if __name__ == '__main__':
    clock()

执行结果：

简单说明一下代码的逻辑：

1、在clock()函数业务逻辑执行完成后，在函数体的最后，启动了一个新的Timer()实例，继续递归执行clock()函数。

2、print()函数通过转义字符\r和end=''，实现在同样的位置进行输出的刷新。

接下来，再简单演示一下失败自动重试的效果，还是直接看代码：

from threading import Timer
import random


def retry_until_success():
    print('开始执行业务逻辑')
    if random.choice([True, False]):
        print('执行成功')
    else:
        print('执行失败，3秒后自动重试')
        timer = Timer(interval=3, function=retry_until_success)
        timer.start()


if __name__ == '__main__':
    retry_until_success()

执行结果：

代码只是简单模拟任务执行失败重试的逻辑，实际使用中，可以将实际业务逻辑封装在一个单独的函数中，成功返回True，否则返回False，则可以通过函数的返回值，判断是否需要自动重试。

总结

本文简单介绍了threading模块中的Timer组件的设计实现，并列举了相关延迟执行、定时调度的实际业务场景。最后通过三个代码示例，分别演示了延迟执行、时间显示刷新、失败自动重试的相关功能实现。

感谢您的拨冗阅读，希望对您有所帮助。