Python实现定时器的5种常用方法与代码示例

基础方案：基于内置模块的定时器实现

第1步：使用threading.Timer控制基础循环

利用Python标准库中的threading.Timer可实现简单的定时递归调用。该方法适合需要动态停止或限制执行次数的场景，核心逻辑是在回调函数中重新实例化Timer对象以维持循环，并通过全局标志位或计数器进行中断控制。

# -*- coding:utf-8 -*-
import threading
import time

cancel_flag = False
def execute_task():
    print("hello world")

def heart_beat():
    print(time.strftime('%Y-%m-%d %H:%M:%S'))
    if not cancel_flag:
        execute_task()
        # 每隔3秒执行一次
        threading.Timer(3, heart_beat).start()

if __name__ == '__main__':
    heart_beat()
    # 15秒后停止定时器
    time.sleep(15)
    cancel_flag = True

第2步：结合datetime计算绝对时间延迟

当需要指定具体的日历时间点（如每天凌晨3点）启动任务时，需先计算当前时间与目标时间的差值（秒），再将该差值作为threading.Timer的等待参数。此方法避免了固定频率可能导致的累积漂移。

# -*- coding:utf-8 -*-
import datetime
import threading

def func():
    print("定时任务已触发")
    # 如需持续循环，可在此重新创建Timer
    timer = threading.Timer(86400, func)
    timer.start()

# 获取当前时间与明天同一时间
now_time = datetime.datetime.now()
nxt_day = now_time + datetime.timedelta(days=1)
# 拼接目标时间（次日03:00:00）
target_str = f"{nxt_day.year}-{nxt_day.month}-{nxt_day.day} 03:00:00"
target_time = datetime.datetime.strptime(target_str, "%Y-%m-%d %H:%M:%S")

# 计算相差秒数
delay_seconds = (target_time - now_time).total_seconds()
print(f"距离目标时间 {delay_seconds:.2f} 秒")

# 创建一次性定时器，到期后自动触发func
scheduler = threading.Timer(delay_seconds, func)
scheduler.start()

轻量级调度：使用schedule库处理周期任务

第3步：配置单一任务的多种执行间隔

schedule是一个语法直观的轻量级定时库，支持秒、分、小时、天甚至特定星期的灵活配置。使用时只需注册任务并配合while循环调用run_pending()即可。该库默认单线程运行，任务按注册顺序依次排队。

# -*- coding:utf-8 -*-
import schedule
import time

def job(text=""):
    print(text, " | I'm working...")

schedule.every().seconds.do(job, "每秒一次")
schedule.every(5).seconds.do(job, "五秒一次")
schedule.every(10).minutes.do(job, "10分钟一次")
schedule.every().hour.do(job, "每小时一次")
schedule.every().day.at("10:30").do(job, "每日10:30执行")
schedule.every(5).to(10).days.do(job, "5至10天内随机执行")
schedule.every().monday.do(job, "每周一执行")
schedule.every().wednesday.at("13:15").do(job, "每周三13:15执行")

while True:
    schedule.run_pending()  # 运行所有已到期的任务
    time.sleep(1)           # 避免空转占用CPU

第4步：编写多任务调度与并发处理策略

当注册多个高频任务时，schedule的串行特性会导致前一个耗时任务阻塞后续任务的精确触发。若业务无法接受延迟，需引入多线程让各定时任务独立执行。

上图展示了未加并发处理时的典型现象：任务1耗时5秒，任务2设定3秒执行一次，实际执行间隔被拉长。修复方案为将job包装进threading.Thread，使调度器仅负责唤醒线程而不等待执行完毕。

# -*- coding:utf-8 -*-
import datetime
import schedule
import threading
import time

def job1():
    print(f"[JOB1] start", datetime.datetime.now())
    time.sleep(5)
    print(f"[JOB1] end", datetime.datetime.now())

def job2():
    print(f"[JOB2] start", datetime.datetime.now())
    time.sleep(3)
    print(f"[JOB2] end", datetime.datetime.now())

# 线程包装函数，交由schedule调度
sched1_task = lambda: threading.Thread(target=job1).start()
sched2_task = lambda: threading.Thread(target=job2).start()

schedule.every(3).seconds.do(sched1_task)
schedule.every(3).seconds.do(sched2_task)

while True:
    schedule.run_pending()
    time.sleep(1)

企业级方案：APScheduler完整定时框架

第5步：初始化调度器并绑定interval、date与cron触发器

对于生产环境，推荐使用APScheduler。它基于事件循环构建，支持内存与数据库持久化，并提供interval（固定间隔）、date（单次绝对时间）和cron（类Linux Crontab复杂规则）三种核心调度模式。

# -*- coding:utf-8 -*-
import time
from apscheduler.schedulers.blocking import BlockingScheduler

def my_job(text="默认任务"):
    print(f"[{text}]", time.strftime('%Y-%m-%d %H:%M:%S'))

sched = BlockingScheduler()

# 1. interval：固定间隔调度
sched.add_job(my_job, 'interval', seconds=3, args=['3秒间隔触发'])

# 2. date：单次绝对时间触发（示例使用相对未来时间演示）
sched.add_job(my_job, 'date', run_date=datetime.datetime.now() + datetime.timedelta(hours=1), args=['1小时后单次执行'])

# 3. cron：复杂周期性调度
# 每月第3个周五的0点至3点执行
sched.add_job(my_job, 'cron', day_of_week='fri', day='3rd', hour='0-3', args=['月度第三个周五cron'])
# 工作日早上5:30执行
sched.add_job(my_job, 'cron', day_of_week='mon-fri', hour=5, minute=30, args=['工作日早间cron'])

try:
    sched.start()  # 阻塞主线程运行调度器
except KeyboardInterrupt:
    print("调度器已停止")

以上就是Python实现定时器的详细内容，更多关于Python定时任务配置与并发处理的资料请关注本站其它相关文章！