Python 怎么实现定时任务和周期任务？threading.Timer 有什么坑？
简化版
threading.Timer(interval, func) 是标准库里最简单的定时器，但要理解它的本质：它就是一个 Thread 子类——每创建一个 Timer 就新起一个线程**，那个线程用 Event.wait(interval) 睡够时间后执行一次函数就结束。所以它是「一次性」的（执行完就没了），想做周期任务只能在函数末尾再创建一个新 Timer——这意味着每个周期都要创建销毁一个线程，高频场景下开销可观；而且用 t.cancel() 只能取消「还没开始执行」的定时器（已经在跑的函数打断不了）。周期任务的第二个坑是「时间漂移」：while True: work(); time.sleep(60) 的实际间隔是「60 秒 + 任务耗时」，如果任务要 5 秒，一天下来就漂移约 2 小时——正确做法是按固定节拍计算下一次的绝对时刻（next_run += interval），或者用 time.monotonic() 累加而不是每次重新 sleep(interval)。第三个坑是异常：定时线程里的函数抛异常会直接杀死这个调度链（Timer 链断了、循环退出了），而且没有任何提示——所有周期任务的函数体都必须用 try/except 兜住。选型上：进程内的简单周期任务用「Event.wait(timeout) 循环」（可被优雅唤醒、无漂移、只用一个线程）；复杂调度用 APScheduler；分布式和运维级的定时用 cron / systemd timer / K8s CronJob / Celery beat——多实例部署时一定要考虑「同一个任务会不会被多个实例同时执行」。核心记忆：Timer 是一次性的一个线程；周期任务要防漂移、防异常吞掉调度、防多实例重复执行。
详细版
四种进程内实现对比：
| 方案 | 线程数 | 能否优雅退出 | 时间漂移 | 适用 |
|---|---|---|---|---|
threading.Timer 递归 | 每周期一个新线程 | cancel()（只能取消未开始的） | 有 | 一次性延时 |
while True: sleep() | 1 | ❌ 要等 sleep 结束 | 有 | 简单脚本 |
Event.wait(timeout) 循环 | 1 | ✅ 立即唤醒 | 可控 | ✅ 推荐 |
sched 模块 | 1（调用方线程） | 需自己处理 | 可控（用绝对时间） | 多任务调度 |
import threading, time, sched, logging
# ① threading.Timer：★一次性★定时器（本质是一个 Thread）
t = threading.Timer(5.0, lambda: print("5 秒后执行"))
t.daemon = True
t.start()
t.cancel() # ★只能取消"还没开始执行"的；已在执行的打不断★
# ② ★递归 Timer 做周期任务（能用但不推荐）★
def periodic(interval, func):
def wrapper():
try:
func()
except Exception:
logging.exception("任务异常") # ★必须兜住，否则链条断了★
finally:
periodic(interval, func) # ★再起一个新线程★
t = threading.Timer(interval, wrapper)
t.daemon = True
t.start()
return t
# ★问题：每个周期创建/销毁一个线程；难以统一取消；有漂移
# ③ ★★推荐：Event.wait 循环（单线程 + 可优雅退出 + 无漂移）★★
class PeriodicTask(threading.Thread):
def __init__(self, interval, func):
super().__init__(daemon=True)
self.interval, self.func = interval, func
self._stop_event = threading.Event()
def run(self):
next_run = time.monotonic() # ★用单调时钟，不受系统改时间影响★
while not self._stop_event.is_set():
try:
self.func()
except Exception:
logging.exception("周期任务异常") # ★吞掉异常，循环继续★
next_run += self.interval # ★固定节拍：不累积任务耗时★
delay = next_run - time.monotonic()
if delay < 0: # ★任务比周期还慢 → 追赶★
next_run = time.monotonic() # 重新对齐（或跳过若干拍）
delay = 0
self._stop_event.wait(delay) # ★能被 stop() 立刻唤醒★
def stop(self):
self._stop_event.set() # ★立即唤醒，不用等睡完★
task = PeriodicTask(60, sync_data)
task.start()
# ...退出时
task.stop(); task.join(timeout=10)
# ④ ★时间漂移的对比★
# ✗ 累积漂移：
while True:
work() # 耗时 5 秒
time.sleep(60) # ★实际周期 = 65 秒 → 一天漂移约 1.85 小时★
# ✓ 固定节拍：
next_run = time.monotonic()
while True:
work()
next_run += 60
time.sleep(max(0, next_run - time.monotonic())) # ★周期恒为 60 秒★
# ⑤ sched 模块（标准库的调度器，单线程）
s = sched.scheduler(time.monotonic, time.sleep)
s.enter(5, priority=1, action=print, argument=("5 秒后",))
s.enterabs(time.monotonic() + 10, 1, print, ("10 秒后",))
s.run() # ★阻塞直到所有任务执行完★（run(blocking=False) 返回下次等待时间）
# ⑥ 生产级：APScheduler
# from apscheduler.schedulers.background import BackgroundScheduler
# sched = BackgroundScheduler()
# sched.add_job(job, "interval", seconds=60, max_instances=1, # ★防重叠★
# coalesce=True, misfire_grace_time=30) # ★防补跑风暴★
# sched.add_job(job, "cron", hour=3, minute=0) # cron 表达式
# sched.start()⚠️ 三个必须记住的坑：①
threading.Timer每次都是一个新线程——它是Thread的子类，构造时创建线程、start()后线程Event.wait(interval)睡够时间执行一次函数就退出。所以「递归 Timer」做周期任务意味着每个周期创建销毁一个线程（每个线程约 8MB 虚拟栈 + 创建开销），秒级周期下就很浪费；而且cancel()只对「尚未开始执行」的定时器有效，正在执行的函数打断不了（Python 无法强制终止线程）。②while True: work(); sleep(N)有累积漂移：实际周期是「N + 任务耗时」，任务耗时 5 秒、周期 60 秒时，一天会少执行约 110 次（漂移近 2 小时）。正确做法是按绝对时刻推进：next_run += interval，然后sleep(next_run - now())——这样任务耗时被「吸收」在周期内，长期不漂移。③ 周期任务里的异常会静默终止整个调度：Timer链断掉、while循环退出，而且通常没有任何日志（子线程的异常不会传播到主线程，只会打印到 stderr 或干脆消失）——所有周期任务的函数体都必须用try/except Exception兜住并记录日志，这是最常见的「定时任务跑着跑着就不跑了」的原因。
完整版教学
一、threading.Timer 的本质
Timer 的实现（简化后就这么几行）：
class Timer(Thread):
def __init__(self, interval, function, args=None, kwargs=None):
Thread.__init__(self)
self.interval = interval
self.function = function
self.finished = Event()
def cancel(self):
self.finished.set() # ★让 wait 立刻返回★
def run(self):
self.finished.wait(self.interval) # ★睡★
if not self.finished.is_set(): # ★没被 cancel 才执行★
self.function(*self.args, **self.kwargs)
self.finished.set()
★ 三个直接推论：
① Timer ★就是一个线程★ → 创建它 = 创建一个线程
② 它是★一次性★的 → 执行完 run() 线程就结束
③ cancel() 只是 set 那个 Event → ★只能阻止"还没开始"的执行★，
已经进入 function 的打断不了
★ 用法要点：
t = threading.Timer(5.0, func, args=(a,), kwargs={"k": v})
t.daemon = True # ★通常要设★，否则主线程退出时会等它
t.start()
t.cancel() # 取消（未开始时有效）
t.is_alive() # 是否还在等待/执行
★ 常见误用：
✗ 用 Timer 做高频周期任务（每秒一次 → 每秒创建销毁一个线程）
✗ 忘了 daemon=True → ★程序退不出去★（主线程等 Timer 线程结束）
✗ 以为 cancel() 能中断正在执行的函数
✗ 创建了大量 Timer 但没有统一管理 → 退出时无法全部取消
★ 什么时候 Timer 是合适的：
✓ ★一次性延时执行★（延迟重试、超时兜底、延迟关闭）
✓ 数量少、周期长（分钟级以上）
✗ 周期任务、高频、需要精确调度 → ★用 Event.wait 循环或 APScheduler★
一个正确的"超时兜底"用法：
def on_timeout():
logging.warning("操作超时，执行兜底")
cleanup()
t = threading.Timer(30, on_timeout); t.daemon = True; t.start()
try:
do_work()
finally:
t.cancel() # ★正常完成就取消兜底★threading.Timer 的实现只有几行：它是 Thread 的子类，run() 里用 Event.wait(interval) 睡够时间，然后检查有没有被 cancel，没有就执行一次函数，线程随即结束。三个直接推论：① 创建一个 Timer 就是创建一个线程；② 它是一次性的（执行完线程就结束）；③ cancel() 只是 set 那个 Event，因此只能阻止「还没开始」的执行——已经进入函数体的打断不了。用法上有两个必须注意的点：通常要设 daemon = True（否则主线程退出时会等它，程序卡住不退），以及别用它做高频周期任务（每秒一次就是每秒创建销毁一个线程）。它真正合适的场景是一次性延时执行：延迟重试、超时兜底（起一个 30 秒的 Timer，正常完成就 cancel()）、延迟关闭。
二、周期任务的四种实现与取舍
① ★递归 Timer★
def loop():
do_work()
threading.Timer(60, loop).start() # ★每次新建线程★
优点：写法简单
缺点：每周期一个新线程；★难以统一 cancel★（要保存最新的 t）；有漂移；
★异常会断链★
② ★while + time.sleep★
while running:
do_work()
time.sleep(60)
优点：最直观
缺点：★停止时要等 sleep 结束★（最坏等一整个周期）；★有累积漂移★
③ ★★while + Event.wait（推荐）★★
stop = threading.Event()
while not stop.is_set():
do_work()
stop.wait(60) # ★stop.set() 能立刻唤醒★
优点：★单线程★、★可立即优雅退出★、可控漂移
→ 这是"进程内周期任务"的标准写法
④ ★sched 模块★
s = sched.scheduler(time.monotonic, time.sleep)
def job():
do_work()
s.enter(60, 1, job) # ★重新安排下一次★
s.enter(0, 1, job)
s.run() # ★阻塞在调用线程里★
优点：支持多任务、优先级、绝对时间（enterabs）
缺点：★单线程串行★（一个任务慢会拖住其他任务）；
run() 阻塞；停止需要自己处理
★ 对比表：
┌────────────────┬────────┬──────────────┬──────────┬──────────────┐
│ 方案 │ 线程数 │ 优雅退出 │ 漂移 │ 适合 │
├────────────────┼────────┼──────────────┼──────────┼──────────────┤
│ 递归 Timer │ ★每周期★│ cancel 最新的 │ 有 │ 一次性延时 │
│ while+sleep │ 1 │ ★要等 sleep★ │ ★累积★ │ 简单脚本 │
│ ★while+Event★ │ 1 │ ★立即★ │ ★可控★ │ ★进程内周期★ │
│ sched │ 1 │ 需自己处理 │ 可控 │ 多任务调度 │
│ APScheduler │ 池 │ shutdown() │ 可控 │ ★复杂调度★ │
└────────────────┴────────┴──────────────┴──────────┴──────────────┘
★ asyncio 版本（如果本来就是异步程序）：
async def periodic(interval, coro, stop: asyncio.Event):
while not stop.is_set():
try:
await coro()
except Exception:
logging.exception("任务异常")
try:
await asyncio.wait_for(stop.wait(), timeout=interval)
break # ★被 set 了 → 退出★
except asyncio.TimeoutError:
pass # ★正常超时 → 下一轮★
→ 或直接 asyncio.sleep(interval) + task.cancel()（★协程的取消是真取消★）周期任务有四种实现。递归 Timer 写法简单但每周期新建线程、难以统一取消、异常会断链。while + time.sleep 最直观，但停止时要等 sleep 结束（最坏等一整个周期）且有累积漂移。while + Event.wait(timeout) 是推荐写法——单线程、能被 stop.set() 立即唤醒实现优雅退出、漂移可控，这是「进程内周期任务」的标准做法。sched 模块支持多任务和优先级、可以用绝对时间（enterabs），但它是单线程串行的（一个任务慢会拖住其他任务）且 run() 会阻塞调用线程。如果本来就是异步程序，用 asyncio 版本更自然——协程的取消是真取消（task.cancel() 会在 await 点抛 CancelledError 并执行 finally）。
三、时间漂移：固定节拍 vs 固定间隔
★ 两种语义要分清：
「固定间隔」：上一次★结束★后再等 N 秒
while True: work(); sleep(60)
→ 实际周期 = 60 + 任务耗时
「固定节拍」：每隔 N 秒执行一次（★不管任务多久★）
next += 60; sleep(next - now)
→ 实际周期恒为 60
★ 漂移的量化（★这个算例要会算★）：
任务耗时 5 秒、期望周期 60 秒：
固定间隔：实际 65 秒一次
一天 86400 秒 → 86400/65 ≈ ★1329 次★
期望 → 86400/60 = 1440 次
★少跑 111 次，相当于漂移约 1.85 小时★
固定节拍：一天恰好 1440 次 ✓
更隐蔽的：任务耗时★不稳定★（1~30 秒波动）
→ 固定间隔下"每次执行的时刻"完全不可预测
→ 日志/监控里看到的执行时间点乱七八糟，难以对账
★ 固定节拍的实现（注意三个细节）：
next_run = time.monotonic() # ★① 用单调时钟★
while not stop.is_set():
run_task()
next_run += interval # ★② 累加，不是重新取 now★
delay = next_run - time.monotonic()
if delay < 0:
# ★③ 任务比周期还慢：怎么办？★
# 策略 A：立刻开始下一次（追赶，可能雪崩）
# 策略 B：跳过错过的拍子，对齐到下一个整拍（★推荐★）
missed = int(-delay // interval) + 1
next_run += missed * interval
logging.warning("任务超时，跳过 %d 次", missed)
delay = next_run - time.monotonic()
stop.wait(delay)
★ 为什么用 time.monotonic() 而不是 time.time()：
time.time() 是★墙上时钟★，会被 NTP 校时、手动改时间、夏令时影响
→ 校时后跳变可能导致：立刻连跑好几次，或者卡住几小时
★ time.monotonic() 单调递增，不受任何时间调整影响★
→ 凡是"测量间隔"的场景，一律用 monotonic
★ 什么时候该用"固定间隔"：
✓ 任务是"轮询下游"且不希望给下游造成固定节拍的压力
✓ 任务耗时本身就是背压信号（下游慢 → 自然放慢频率）
✗ 需要"每分钟第 0 秒执行"这类对齐要求 → ★必须固定节拍★
★ 对齐到整点/整分（cron 语义）：
import datetime
def next_minute_boundary():
now = datetime.datetime.now()
return (now.replace(second=0, microsecond=0)
+ datetime.timedelta(minutes=1))
★ 这类"日历语义"的调度要用 time.time()/datetime（因为要对齐真实时钟），
但要处理 ★夏令时、时区变更、系统改时间★ → 直接用 APScheduler 的 cron 触发器更稳时间漂移的根源是混淆了两种语义：「固定间隔」（上一次结束后再等 N 秒）和「固定节拍」（每隔 N 秒执行一次，不管任务多久）。量化一下就很直观：任务耗时 5 秒、期望周期 60 秒时，固定间隔的实际周期是 65 秒——一天少跑 111 次，相当于漂移近 2 小时；任务耗时不稳定时，执行时刻更是完全不可预测。固定节拍的实现有三个细节：① 用 time.monotonic() 而不是 time.time()（墙上时钟会被 NTP 校时、手动改时间、夏令时影响，跳变后可能立刻连跑好几次或卡住几小时；凡是测量间隔一律用单调时钟）；② 累加 next_run += interval 而不是每次重新取 now；③ 想清楚「任务比周期还慢」时的策略——是立刻追赶（可能雪崩）还是跳过错过的拍子对齐到下一个整拍（推荐）。至于「每分钟第 0 秒执行」这类日历语义的对齐，必须用真实时钟，但要处理夏令时和时区变更，直接用 APScheduler 的 cron 触发器更稳妥。
四、异常、重叠与优雅退出
★ 坑一：异常静默杀死调度★
✗ def job():
data = fetch() # ★某次抛异常★
save(data)
while not stop.is_set():
job() # ★异常向上传播 → 循环退出 → 定时任务再也不跑了★
stop.wait(60)
现象：★定时任务"跑着跑着就不跑了"，而且没有任何日志★
（子线程的未捕获异常只会打到 stderr，很多部署环境根本看不到）
✓ 正确：
while not stop.is_set():
try:
job()
except Exception:
logging.exception("定时任务失败") # ★记录但不中断循环★
stop.wait(60)
★ 注意用 except Exception 而不是 except BaseException
（否则会吞掉 KeyboardInterrupt/SystemExit）
★ 坑二：任务重叠（上一次还没跑完，下一次又开始）★
固定节拍 + 任务偶尔变慢 → 两次执行同时进行
→ 后果：重复写数据、并发访问同一资源、连接数翻倍、数据错乱
✓ 方案 A：加锁跳过
lock = threading.Lock()
def job():
if not lock.acquire(blocking=False):
logging.warning("上一次还在跑，本次跳过")
return # ★跳过而不是排队★
try: work()
finally: lock.release()
✓ 方案 B：APScheduler 的 max_instances=1（★内置支持★）
✓ 方案 C：改成"串行循环"（做完再等，即固定间隔语义）
★ 坑三：优雅退出★
✗ time.sleep(3600) → ★退出时最坏要等一小时★
✓ stop.wait(3600) → ★stop.set() 立刻唤醒★
✗ daemon=True 一了百了 → ★任务执行到一半被强杀★（数据不一致）
✓ 非 daemon + stop 标志 + join(timeout)
完整退出流程：
signal 处理器 → stop.set()
→ 各周期任务线程从 wait 中醒来、退出循环
→ 主线程 join(timeout) 等它们收尾
→ 超时则记录告警
★ 坑四：多实例重复执行（★分布式部署的头号问题★）★
服务部署了 3 个副本，每个都起了同样的周期任务
→ ★同一个任务每分钟被执行 3 次★
→ 后果：重复发邮件、重复扣款、重复写入
✓ 方案：
① ★分布式锁★（Redis SETNX + TTL / etcd / 数据库唯一约束）
if redis.set("lock:job", instance_id, nx=True, ex=55):
run_job()
② ★交给外部调度★：K8s CronJob / cron / Celery beat（★只有一个调度器★）
③ ★主从选举★：只有 leader 执行定时任务
④ 任务本身★幂等★（最根本的解法：重复执行也无害）
★ 注意分布式锁的 TTL 要 > 任务最长耗时，否则锁过期后别人也会执行周期任务有四个必须处理的坑。① 异常静默杀死调度：函数抛异常会让循环退出、Timer 链断掉，而且通常没有任何日志（子线程的未捕获异常只会打到 stderr），表现为「定时任务跑着跑着就不跑了」——必须用 try/except Exception 兜住并记录（注意用 Exception 而不是 BaseException，否则会吞掉 KeyboardInterrupt）。② 任务重叠：固定节拍下任务偶尔变慢就会出现两次执行同时进行，导致重复写数据、连接翻倍——解法是用非阻塞锁跳过本次或用 APScheduler 的 max_instances=1。③ 优雅退出：time.sleep(3600) 会让退出最坏等一小时，必须用 stop.wait(3600)；也不要用 daemon=True 一了百了（任务会被执行到一半强杀）。④ 多实例重复执行是分布式部署的头号问题：3 个副本各起一份周期任务，同一个任务每分钟被执行 3 次——解法是分布式锁（注意 TTL 要大于任务最长耗时）、交给外部调度器、主从选举，或者把任务做成幂等的（最根本）。
五、生产级方案选型
★ 按"调度器在哪"分成三类：
① ★进程内调度★（调度逻辑在你的应用里）
threading.Timer / Event 循环 / sched / ★APScheduler★
✓ 优点：部署简单、能访问应用内的状态和连接
✗ 缺点：★应用重启就丢★、★多副本会重复执行★、和业务共享资源
② ★外部调度器触发★
cron / systemd timer / K8s CronJob → 起一个新进程执行
✓ 优点：★调度和执行解耦★、天然单实例、崩溃不影响、有系统级日志
✗ 缺点：每次要重新启动进程（初始化开销）、传参和状态共享麻烦
★ K8s CronJob 要注意：concurrencyPolicy（Allow/Forbid/Replace）、
startingDeadlineSeconds、successfulJobsHistoryLimit
③ ★分布式任务队列★
Celery beat / Dramatiq / RQ-Scheduler / Temporal / Airflow
✓ 优点：★调度 + 分发 + 重试 + 监控★一体，适合复杂依赖和大规模
✗ 缺点：引入 broker（Redis/RabbitMQ）等基础设施
★ Celery beat 本身★必须单实例★（否则重复投递），常配合 redbeat 做高可用
★ APScheduler（进程内的事实标准）：
from apscheduler.schedulers.background import BackgroundScheduler
sched = BackgroundScheduler(timezone="Asia/Shanghai") # ★显式时区★
sched.add_job(job, "interval", minutes=5,
id="sync", replace_existing=True,
max_instances=1, # ★禁止重叠执行★
coalesce=True, # ★错过多次只补跑一次（防风暴）★
misfire_grace_time=60) # ★超过 60 秒就不补跑了★
sched.add_job(job2, "cron", hour=3, minute=30) # cron 语义
sched.add_job(job3, "date", run_date="2026-09-01 10:00") # 一次性
sched.start()
...
sched.shutdown(wait=True) # ★优雅关闭★
★ 三个关键参数（面试可讲）：
max_instances 同一个 job 最多几个实例并行（★默认 1★）
coalesce 积压的多次触发★合并成一次★
（服务停机 1 小时后重启，避免瞬间补跑 60 次）
misfire_grace_time 错过多久之内还补跑，超过就放弃
★ 选型决策：
┌──────────────────────────────┬──────────────────────────┐
│ 场景 │ 推荐 │
├──────────────────────────────┼──────────────────────────┤
│ 进程内、简单、单实例 │ ★Event.wait 循环★ │
│ 进程内、多任务、cron 语义 │ ★APScheduler★ │
│ 独立脚本、运维可见 │ ★cron / systemd timer★ │
│ 容器化、要天然单实例 │ ★K8s CronJob★ │
│ 分布式、要重试和监控 │ ★Celery beat / Airflow★ │
│ 复杂依赖、工作流 │ Airflow / Prefect / Temporal│
└──────────────────────────────┴──────────────────────────┘
★ 一个务实的建议：
★能交给外部调度器就别写在应用里★——
应用内的定时任务会随发布重启而中断、多副本会重复、
出问题时和业务日志混在一起难排查。生产级方案按「调度器在哪」分三类。① 进程内调度（Event 循环、APScheduler）——部署简单、能访问应用状态，但应用重启就丢、多副本会重复执行。② 外部调度器触发（cron、systemd timer、K8s CronJob）——调度和执行解耦、天然单实例、有系统级日志，代价是每次要重启进程。③ 分布式任务队列（Celery beat、Airflow）——调度、分发、重试、监控一体，代价是引入 broker（注意 Celery beat 本身必须单实例）。APScheduler 是进程内的事实标准，三个关键参数要会说：max_instances（禁止重叠执行）、coalesce（积压的多次触发合并成一次，避免停机一小时后瞬间补跑 60 次）、misfire_grace_time（错过多久之内还补跑）。最后一个务实建议：能交给外部调度器就别写在应用里——应用内的定时任务会随发布重启而中断、多副本会重复执行、出问题时日志还和业务混在一起。
六、实践清单
★ 写周期任务的检查清单：
□ ★函数体用 try/except Exception 兜住★（否则异常会杀死调度）
□ ★用 Event.wait(timeout) 而不是 time.sleep★（能优雅退出）
□ ★用 time.monotonic() 计算间隔★（不受系统改时间影响）
□ 固定节拍：next_run += interval（★不要 sleep(interval)★）
□ 想清楚"任务比周期慢"时的策略（跳过 or 追赶）
□ ★防重叠★：非阻塞锁跳过 / max_instances=1
□ ★防多实例★：分布式锁 / 外部调度 / 幂等设计
□ 线程非 daemon + stop 标志 + join(timeout)
□ ★每次执行记日志★（开始、结束、耗时、结果）——否则出问题查不到
□ 加监控：★上次成功执行的时间★（超过阈值告警 = 发现"悄悄不跑了"）
★ 时间相关的通用纪律：
测量间隔 / 超时 → ★time.monotonic()★（单调，不受校时影响）
记录"何时发生" → time.time() / datetime（★带时区★）
日历语义（每天 3 点）→ datetime + 时区库；★注意夏令时★
★永远不要用 datetime.now() 相减来测耗时★
★ 常见故障与原因：
┌────────────────────────────┬────────────────────────────────┐
│ 现象 │ 原因 │
├────────────────────────────┼────────────────────────────────┤
│ 任务跑着跑着就不跑了 │ ★异常没捕获，循环退出★ │
│ 执行时间越来越晚 │ ★累积漂移（sleep 固定间隔）★ │
│ 同一任务被执行多次 │ ★多副本 / 重叠执行★ │
│ 程序退不出去 │ ★非 daemon 线程卡在 sleep★ │
│ 停服后重启瞬间跑了几十次 │ ★补跑风暴（没设 coalesce）★ │
│ 改了系统时间后行为异常 │ ★用了 time.time() 测间隔★ │
│ 任务执行了但没效果 │ 多实例互相覆盖 / 事务未提交 │
└────────────────────────────┴────────────────────────────────┘
★ 一个完整的生产级模板：
class PeriodicJob(threading.Thread):
def __init__(self, name, interval, func):
super().__init__(name=name, daemon=False) # ★非 daemon★
self.interval, self.func = interval, func
self._stop = threading.Event()
self._lock = threading.Lock() # ★防重叠★
def run(self):
nxt = time.monotonic()
while not self._stop.is_set():
if self._lock.acquire(blocking=False):
t0 = time.monotonic()
try:
self.func()
logging.info("%s 完成，耗时 %.2fs", self.name,
time.monotonic() - t0)
except Exception:
logging.exception("%s 失败", self.name) # ★不中断★
finally:
self._lock.release()
else:
logging.warning("%s 上次未完成，跳过", self.name)
nxt += self.interval # ★固定节拍★
d = nxt - time.monotonic()
if d < 0:
nxt = time.monotonic(); d = 0 # 重新对齐
self._stop.wait(d) # ★可唤醒★
def stop(self):
self._stop.set()最后是实践清单。写周期任务必查的十条里，最容易漏的是：函数体用 try/except Exception 兜住、用 Event.wait 而不是 time.sleep、用 time.monotonic() 计算间隔、防重叠和防多实例、以及加监控：记录「上次成功执行的时间」，超过阈值就告警——这是发现「任务悄悄不跑了」的唯一有效手段。时间相关还有一条通用纪律：测量间隔和超时一律用 time.monotonic()（单调、不受 NTP 校时影响），记录「何时发生」才用 time.time()/datetime，永远不要用 datetime.now() 相减来测耗时。那张故障对照表很实用：跑着跑着不跑了 = 异常没捕获、执行时间越来越晚 = 累积漂移、重启瞬间跑了几十次 = 补跑风暴（没设 coalesce）、程序退不出去 = 非 daemon 线程卡在 sleep。
记忆钩子：「★threading.Timer 本质就是一个 Thread 子类★：run() 里 Event.wait(interval) 睡够就执行一次函数然后线程结束——所以①★创建一个 Timer 就是创建一个线程★②它是★一次性★的（递归做周期任务＝每周期新建销毁一个线程）③★cancel() 只能取消『还没开始执行』的★，已经在跑的打不断（Python 无法强制终止线程）；还要记得设 daemon=True，否则程序退不出去。它真正适合的是★一次性延时★（超时兜底、延迟重试）。★周期任务的四个坑★：①★异常静默杀死调度★——函数抛异常会让循环退出/Timer 链断掉且★通常没有日志★（子线程异常只打到 stderr），这就是『定时任务跑着跑着就不跑了』，必须 try/except Exception 兜住（★别用 BaseException，会吞掉 KeyboardInterrupt★）；②★时间漂移★——
work(); sleep(60)的实际周期是『60+任务耗时』，任务 5 秒时一天★少跑 111 次、漂移近 2 小时★，要改成★固定节拍 next_run += interval★，并且★用 time.monotonic() 而不是 time.time()★（墙上时钟会被 NTP 校时/改时间/夏令时跳变影响）；③★任务重叠★（上次没跑完下次又开始）→ 非阻塞锁跳过或 APScheduler 的 max_instances=1；④★多实例重复执行★（3 个副本各跑一份）→ 分布式锁（★TTL 要大于任务最长耗时★）/ 交给外部调度 / ★任务做成幂等★。★推荐写法是『while not stop.is_set(): try: work() except: log; stop.wait(delay)』★——单线程、★stop.set() 能立刻唤醒实现优雅退出★（time.sleep 最坏要等一整个周期）、漂移可控。选型：进程内简单任务用 Event 循环、多任务和 cron 语义用 ★APScheduler★（三个关键参数 max_instances 防重叠、★coalesce 防重启后的补跑风暴★、misfire_grace_time）、容器化要天然单实例用 ★K8s CronJob★、分布式要重试监控用 Celery beat/Airflow。★务实建议：能交给外部调度器就别写在应用里★（应用内的会随发布重启中断、多副本会重复）。监控上必须记录★『上次成功执行的时间』★并告警，否则任务悄悄停了都不知道。」
七、常见误区与追问
- 误区：
threading.Timer是一个轻量的定时器，可以大量创建。 它是Thread的子类——每创建一个 Timer 就是创建一个操作系统线程（每个线程约 8MB 虚拟栈空间，加上创建和销毁的开销）。用「递归 Timer」实现秒级周期任务，就是每秒创建销毁一个线程；同时管理几百个定时器就是几百个线程。而且它是一次性的：run()执行完函数后线程就结束了。正确的替代是：周期任务用「一个线程 +Event.wait(timeout)循环」（单线程搞定，还能被立刻唤醒）；需要管理大量定时任务时用sched模块（单线程调度多个任务）或 APScheduler（内部用线程池）；异步程序里直接用asyncio.sleep+create_task（协程开销远小于线程）。 - 误区：调用
timer.cancel()就能停止定时器的执行。 只能取消还没开始执行的定时器。看Timer的实现就清楚了：cancel()做的是self.finished.set()，让run()里的finished.wait(interval)立刻返回，然后通过if not self.finished.is_set()判断跳过函数调用。所以：如果函数已经开始执行，cancel()毫无作用——Python 没有强制中断线程的机制，那个函数会一直跑到自然结束。这在「超时兜底」场景要特别注意：t.cancel()只保证「还没触发的不会触发」，不保证「正在跑的会停下」。需要中断正在执行的任务，只能靠协作式取消（任务内部检查Event标志）。 - 误区：
while True: work(); time.sleep(60)就是每分钟执行一次。 实际周期是 「60 秒 + 任务耗时」。任务耗时 5 秒时实际是 65 秒一次——一天下来少执行约 111 次，累计漂移近 2 小时；如果任务耗时波动（1~30 秒），执行时刻会完全不可预测，日志和监控都难以对账。这叫累积漂移，因为每次的误差都叠加到后面。正确做法是按绝对时刻推进（固定节拍）：维护一个next_run变量，每轮next_run += interval，然后sleep(next_run - now())——任务耗时被「吸收」在周期内，长期不漂移。同时要想清楚「任务比周期还慢」时怎么办：立刻追赶（可能雪崩）还是跳过错过的拍子对齐到下一拍（推荐，并记录告警）。 - 误区：定时任务的函数抛异常，最多这一次不执行，下次还会正常跑。 整个调度会永久停止。
while循环里的异常会向上传播导致循环退出；递归 Timer 的异常会让「再创建下一个 Timer」那行代码执行不到，链条彻底断裂。更糟的是通常没有任何提示：子线程的未捕获异常不会传播到主线程，只会由threading.excepthook打印到 stderr——而很多部署环境（后台服务、容器）根本没人看 stderr，或者它被重定向丢弃了。表现就是「定时任务跑着跑着就不跑了」，往往过了几天才被发现。所有周期任务的函数体都必须用try/except Exception包住并记日志（用Exception而不是BaseException，否则会吞掉KeyboardInterrupt/SystemExit），并且**监控「上次成功执行时间」**才能及时发现。 - 误区：服务部署了多个副本，定时任务也会自动只执行一次。 每个副本都会独立执行——3 个副本的「每分钟同步一次」实际上是每分钟执行 3 次，后果可能是重复发邮件、重复扣款、重复写入导致数据错乱，或者三个实例并发操作同一资源产生竞态。这是分布式部署下定时任务的头号问题，且在单实例的测试环境完全不会暴露。四种解法：① 分布式锁（
redis.set(key, id, nx=True, ex=TTL)，注意 TTL 必须大于任务最长耗时，否则锁过期后别的实例也会开始执行）；② 交给外部调度器（K8s CronJob、cron、Celery beat —— 调度器只有一个）；③ 主从选举（只有 leader 执行）；④ 把任务做成幂等的（最根本，重复执行也无害）。 - 追问：为什么测量时间间隔要用
time.monotonic()而不是time.time()？ 因为time.time()是墙上时钟（wall clock），它会被 NTP 校时、管理员手动改系统时间、夏令时切换、虚拟机快照恢复等操作改变，甚至可能倒退。用它计算周期任务的间隔会出现两类诡异故障：时钟向前跳（比如校准时快进了 1 小时）→next_run - now()变成负数 → 任务瞬间连续执行几十次；时钟向后跳 → 计算出的等待时间变得很长 → 任务卡住几小时不执行。而time.monotonic()是单调时钟，从某个不确定的起点开始单调递增，不受任何时间调整影响（它的绝对值没有意义，只能用于相减）。规则很清晰：「测量间隔、超时、耗时」一律用monotonic()；「记录事件发生在什么时刻」才用time.time()/datetime（并且要带时区）。至于「每天凌晨 3 点执行」这类日历语义的调度，确实需要真实时钟，但要处理夏令时和时区变更——直接用 APScheduler 的 cron 触发器比自己算更稳妥。 - 追问：APScheduler 的
coalesce和misfire_grace_time分别解决什么问题？ 两者都用于处理「错过的触发」（misfire）——比如服务停机维护了 1 小时，而某个任务是每分钟执行一次，重启后调度器发现有 60 次触发被错过了。coalesce=True（合并） 表示「把这些积压的触发合并成一次执行」，避免服务刚启动就瞬间跑 60 次（这种「补跑风暴」经常直接把刚恢复的服务再次打垮，是很典型的二次故障）。misfire_grace_time=N表示「错过的触发如果超过 N 秒就干脆放弃，不再补跑」——对于「每分钟同步一次」这类任务，一小时前该跑的那次现在补跑已经没有意义了。两者通常一起用：coalesce=True, misfire_grace_time=60，含义是「错过 60 秒以内的合并成一次补跑，超过就跳过」。配合max_instances=1（同一 job 最多一个实例在跑，防止重叠），这三个参数构成了 APScheduler 生产配置的核心。 - 追问：定时任务应该写在应用里还是交给外部调度器？ 能交给外部就别写在应用里，理由有四个：① 生命周期解耦——应用内的定时任务会随着发布、重启、扩缩容而中断或重复，而 cron/K8s CronJob 的调度不受应用部署影响；② 天然单实例——外部调度器只有一个，不会出现「N 个副本各跑一份」的问题（应用内则必须自己加分布式锁）；③ 故障隔离——定时任务把内存吃爆或死循环时，不会拖垮正在服务用户请求的应用进程；④ 可观测性——独立的执行记录、退出码、系统级日志，出问题时不用在海量业务日志里捞。什么时候适合放在应用内：任务非常轻量且高频（每几秒一次，起进程的开销不划算）、需要访问应用的内存状态（本地缓存刷新、连接池健康检查、指标上报）、或者部署环境没有外部调度能力。折中方案是在应用内暴露一个 HTTP 端点，由外部调度器定时调用——既保留了访问应用状态的能力，又把调度职责交给了外部（记得给这个端点加鉴权和防重入）。
八、加强记忆
threading.Timer 本质就是一个 Thread 子类：run() 里用 Event.wait(interval) 睡够时间就执行一次函数、然后线程结束——所以 ① 创建一个 Timer 就是创建一个线程、② 它是一次性的（递归做周期任务等于每周期新建销毁一个线程）、③ cancel() 只能取消「还没开始执行」的（已经在跑的打不断，因为 Python 无法强制终止线程）；还要记得设 daemon=True，否则程序退不出去。它真正适合的是一次性延时（超时兜底、延迟重试）。周期任务有四个坑：① 异常静默杀死调度——函数抛异常会让循环退出或 Timer 链断裂，且通常没有任何日志（子线程异常只打到 stderr），这正是「定时任务跑着跑着就不跑了」的原因，必须 try/except Exception 兜住（别用 BaseException，会吞掉 KeyboardInterrupt）；② 时间漂移——work(); sleep(60) 的实际周期是「60 + 任务耗时」，任务 5 秒时一天少跑 111 次、漂移近 2 小时，要改成固定节拍 next_run += interval，并且用 time.monotonic() 而不是 time.time()（墙上时钟会被 NTP 校时、改时间、夏令时跳变影响，导致瞬间连跑或卡住几小时）；③ 任务重叠（上次没跑完下次又开始）→ 用非阻塞锁跳过或 APScheduler 的 max_instances=1；④ 多实例重复执行（3 个副本各跑一份）→ 分布式锁（TTL 必须大于任务最长耗时）、交给外部调度、或把任务做成幂等的。推荐写法是「while not stop.is_set(): try: work() except: log; stop.wait(delay)」——单线程、stop.set() 能立刻唤醒实现优雅退出（time.sleep 最坏要等一整个周期）、漂移可控。选型上：进程内简单任务用 Event 循环、多任务和 cron 语义用 APScheduler（三个关键参数：max_instances 防重叠、coalesce 防重启后的补跑风暴、misfire_grace_time 控制补跑窗口）、容器化且要天然单实例用 K8s CronJob、分布式且要重试监控用 Celery beat / Airflow。务实建议：能交给外部调度器就别写在应用里（应用内的会随发布重启中断、多副本会重复、故障还会拖累业务进程）。最后，监控上必须记录**「上次成功执行的时间」并设置告警**——否则任务悄悄停了都没人知道。