Python中利用threading.Timer实现定时与周期性任务调度

Python定时器功能实现原理

在Python中，threading.Timer 是实现定时任务的核心工具。它本质上是 Thread 类的派生类，专门用于在指定的延迟时间后调用一个函数。由于 Timer 默认只执行一次，若需实现周期性（循环）调用，必须在被调用的函数内部再次创建并启动一个新的 Timer 对象，从而形成闭环调度。

基础定时器示例

以下代码演示了如何创建一个简单的定时器。该示例设置了1秒的延迟间隔，并在主线程中通过 time.sleep 保持运行，以便观察后台定时任务的执行情况。

#!/usr/bin/env python
from threading import Timer
import time

timer_interval = 1

def delayrun():
    print('running')

t = Timer(timer_interval, delayrun)
t.start()

while True:
    time.sleep(0.1)
    print('main running')

生产级周期性任务调度实现

在实际应用中，定时任务往往需要处理复杂的业务逻辑（如计算下次执行时间、处理异常退出等）。以下代码展示了一个完整的周期性调度方案，结合了 signal 模块处理进程退出信号，以及 datetime 模块计算精确的下次执行间隔。

核心逻辑解析

1. 调度入口：定义 schedule_update 函数，创建 threading.Timer 对象并设置为守护线程（setDaemon(True)），确保主线程退出时子线程也随之终止。
2. 执行函数：在 event_func 中执行具体业务逻辑（如打印当前时间、调用更新函数），随后计算下一次执行的延迟时间。
3. 递归调度：在 event_func 末尾，根据计算出的延迟时间创建新的 Timer 并启动，实现周期性循环。
4. 安全退出：通过 signal.signal 捕获 SIGINT 和 SIGTERM 信号，调用 sys.exit() 安全终止程序。

import time
import sys
import signal
import datetime
import threading

# 定时器调度入口
def schedule_update():
    t = threading.Timer(0, event_func)
    t.setDaemon(True)
    t.start()

# 执行函数
def event_func():
    now_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    print(now_time)
    # exec_update()
    # update_openvas_dbs_from_cache()
    interval_time = delay_time()
    t = threading.Timer(interval_time, event_func)
    t.setDaemon(True)
    t.start()

# 计算下次执行的时间间隔
def delay_time():
    # 当前时间
    now_time = datetime.datetime.now()
    # 明天的时间
    next_time = now_time + datetime.timedelta(days=+1)
    next_year = next_time.date().year
    next_month = next_time.date().month
    next_day = next_time.date().day
    # 获取明天 00:00:00
    next_time = datetime.datetime.strptime(str(next_year)+"-"+str(next_month)+"-"+str(next_day)+" 00:00:00", "%Y-%m-%d %H:%M:%S")
    # 计算秒数差
    delay_time = (next_time - now_time).total_seconds()
    return delay_time

def quit_sys(signum, frame):
    sys.exit()

# 接收C
if __name__ == "__main__":
    try:
        signal.signal(signal.SIGINT, quit_sys)
        signal.signal(signal.SIGTERM, quit_sys)
        schedule_update()
        print("schedule_update server starting up...\nHit Ctrl-C to quit.\n")
        while 1:
            time.sleep(1)
    except Exception as e:
        print(e)

关键注意事项

在实现周期性定时器时，需注意以下几点：

• 守护线程设置：务必调用 t.setDaemon(True)，否则主线程退出后，后台的 Timer 线程仍会保持运行，导致程序无法正常退出。
• 异常处理：在 event_func 中应包含适当的异常捕获机制，防止因业务逻辑报错导致定时器链断裂。
• 时间计算精度：使用 datetime 模块计算时间差时，注意时区与夏令时可能带来的影响，确保 total_seconds() 返回值的准确性。

以上就是Python中利用threading.Timer实现定时与周期性任务调用的详细内容，更多关于Python多线程编程与后台任务处理的资料请关注本站其它相关文章！