使用Python快速实现一个高级定时任务

Python定时方法可以适用的场景：

第一种，用Python定期执行一些后台任务，比如说清理缓存，更新数据，发送通知等；

第二种，用Python定时触发一些事件，比如启动或停止某些服务，执行某些命令，调用某些接口等等；

第三种，通过Python实现一些定时功能，比如说闹钟，倒计时，计时器任务等等。

最后，也可以通过Python模拟一些异步或并发的行为，比如说模拟用户的请求，测试程序的性能，生成一些随机数据等等多种方面。

说的再多，不如直接通过案例来说明。

比如说，现在我们通过Python来创建重复执行的定时任务。如果我们想要创建一个重复执行的定时器任务，这里可以选择使用threading模块的Timer类，并在回调函数中重新创建并启动一个新的定时器任务。

示例如下所示，我们用下面的代码创建了一个每隔5秒打印一条消息的定时任务：

import threading


def demo():
    print("Hello, Copilot!")
    # 创建并启动一个新的定时器
    timer = threading.Timer(5, demo)
    timer.start()


# 创建并启动第一个定时器
timer = threading.Timer(5, demo)
timer.start()

总的来说，就是先调用demo实现启动，如何在demo内部实现再次调用，这样就可以减少非必要的代码冗余。

那么，有同学可能想问了，那我们如何在执行过程中去取消它或者是修改这个定时任务呢？

如果说，我们想要取消或修改一个已经创建的定时任务，可以使用Timer类的cancel()方法或者sched模块的scheduler类的cancel()方法去实现。

举个例子，我们可以通过下面的代码，创建一个在8秒后打印一条消息的定时任务，但是我们在5秒之后就自动把它取消掉：

import threading
import time


def demo():
    print("Hello, Copilot!")


# 创建一个定时任务
timer = threading.Timer(8, demo)
timer.start()


# 等待 5 秒
time.sleep(5)


# 取消定时器
timer.cancel()

除此之外，我们还需要注意到一种情况。

如果我们的定时任务中，它的回调函数中发生了异常或者错误提示，这个时候可以使用try...except语句来捕获并处理它们。具体需要怎么做呢？

我们再举个例子，可以通过下面的代码创建一个定时任务，它的回调函数中故意抛出了一个异常错误，并在except语句中打印了异常信息，如下所示：

import threading


def demo():
    # 抛出一个异常
    raise Exception("Something went wrong!")


# 创建一个定时任务
timer = threading.Timer(5, demo)
timer.start()


# 捕获并处理异常
try:
    timer.join()
except Exception as error:
    print("Error:", error)