qidao123.com技术社区-IT企服评测·应用市场

标题: 一文速通 Python 并行计算：05 Python 多线程编程-线程的定时运行 [打印本页]

作者: 鼠扑 时间: 2025-3-30 22:31
标题: 一文速通 Python 并行计算：05 Python 多线程编程-线程的定时运行
一文速通 Python 并行计算：05 Python 多线程编程-线程的定时运行

摘要：

本文主要讲述了 Python 如何实现定时任务，主要有四种方式：通过 threading.Timer 类、通过 sched 模块、通过 threading.Event 事件机制、通过 APScheduler 第三方库，结合例程分析了各自使用场景和基本方法。

关于我们更多先容可以查看云文档：Freak 嵌入式工作室云文档，或者访问我们的 wiki：****https://github.com/leezisheng/Doc/wik

正文

1. 使用 threading.Timer 类

Timer 继续子 Thread 类，是 Thread 的子类，也是线程类，具有线程的本领和特征。这个类用来定义多久执行一个函数。它的实例是能够耽误执行目标函数的线程，在真正执行目标函数之前，都可以 cancel 它。

threading.Timer(interval, function, args=None, kwargs=None)

复制代码

创建一个定时器，在经过 interval 秒的间隔事件后，将会用参数 args 和关键字参数 kwargs 调用 function。如果 args 为 None，则会使用一个空列表。如果 kwargs 为 None，则会使用一个空字典。
与线程一样，定时器也是通过调用其 Timer.start 方法来启动的。定时器可以通过调用 cancel() 方法来停止（在其动作开始之前）。定时器在执行其行动之前要等候的时间间隔可能与用户指定的时间间隔不完全相同。
以下为定时闹钟的示例代码:

import threading
def thread_Timer():
print("该起床啦...5秒之后再次呼叫你起床...")
_# 声明全局变量_
global t1
_# 创建并初始化线程_
t1 = threading.Timer(5, thread_Timer)
_# 启动线程_
t1.start()
if __name__ == "__main__":
_# 创建并初始化线程_
t1 = threading.Timer(5, thread_Timer)
_# 启动线程_
t1.start()

复制代码

在主线程创建了线程定时器，在 5 秒之后执行 thread_Timer 线程函数，而在 thread_Timer 函数结束的时间，又设置了定时器线程 thread_Timer，这就完成了一个递归的操纵，间隔 5 秒重复定时任务！
注意，以上代码在 Python 3.12 版本中会出现莫名其妙错误，具体情况和办理方法可以看：https://github.com/python/cpython/issues/113964

2. 使用 sched 模块

sched 是 Python 标准库中的调度模块，适用于 一次性任务调度，基于 time.sleep 进行任务执行。

import sched
import time
# 创建调度器
scheduler = sched.scheduler(time.time, time.sleep)
def task():
print(f"任务执行了！当前时间：{time.strftime('%X')}")
# 5 秒后执行 task
scheduler.enter(5, 1, task, ())
print("任务已安排，主线程继续运行...")
scheduler.run() # 开始调度

复制代码

以上代码中 scheduler.enter(5, 1, task, ()) 在 5 秒后运行 task 函数，scheduler.run() 启动调度器，会阻塞主线程直到任务执行完毕。
它适用于单次任务，但不支持周期性任务，同时不会创建新线程，任务执行时会阻塞主线程；调度精度受 time.sleep 限定，可能不太精准。
3. 使用 threading.Event 类

threading.Event 是一种线程同步机制，适用于周期性任务，可以在任务运行期间动态停止任务。

import threading
import time
def periodic_task(interval, stop_event):
while not stop_event.is_set(): # 只要 stop_event 没有被设置，就一直循环
print(f"执行周期性任务... 当前时间：{time.strftime('%X')}")
stop_event.wait(interval) # 等待指定时间间隔
# 创建停止事件
stop_event = threading.Event()
# 启动定时任务，每 5 秒执行一次
thread = threading.Thread(target=periodic_task, args=(5, stop_event))
thread.start()
# 运行一段时间后停止
time.sleep(20)
stop_event.set() # 触发停止事件
thread.join() # 等待线程结束
print("定时任务已停止。")

复制代码

threading.Thread 启动一个独立的线程运行 periodic_task，不会阻塞主线程，stop_event.wait(interval) 控制间隔时间，并可以随时停止任务，stop_event.set() 终止任务，thread.join() 等候任务结束。
它适用于周期性任务，比 sched 更机动，同时任务可以随时终止，避免 while True 死循环，而且不阻塞主线程，适合后台定时任务；但由于 threading 受 GIL（全局解释器锁）影响，不适用于 CPU 密集型任务。
4.使用 APScheduler 库

APScheduler 是一个功能强盛的 Python 定时任务库，支持一次性任务、固定间隔任务、定时任务（Cron）等，适用于复杂的、需要精确调度的定时任务（如每天 8:00 运行），支持多种调度方式（间隔执行、特定时间执行、Cron 表达式）。
起首通过 pip 工具安装 APScheduler 库：

pip install apscheduler

复制代码

以下是定时执行任务的例程：

from apscheduler.schedulers.background import BackgroundScheduler
import time
def job():
print(f"定时任务执行了！当前时间：{time.strftime('%X')}")
scheduler = BackgroundScheduler()
scheduler.add_job(job, 'interval', seconds=5) # 每 5 秒执行一次
scheduler.start()
try:
while True:
time.sleep(1) # 主线程保持运行，否则任务会终止
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown() # 关闭调度器

复制代码

下面是在特定时间执行任务的例程：

from apscheduler.schedulers.blocking import BlockingScheduler
import datetime
scheduler = BlockingScheduler()
def job():
print("特定时间任务执行！")
# 在 2025-03-30 18:30:00 执行
run_time = datetime.datetime(2025, 3, 30, 18, 30, 0)
scheduler.add_job(job, 'date', run_date=run_time)
print("任务已安排...")
scheduler.start() # 阻塞主线程

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/)