对账任务要求每 90 秒拉一次上游文件，cron 只能给到分钟级；更麻烦的是服务部署了三个副本，谁都不该多跑一遍。
这类需求不需要引入 Celery 或 APScheduler。threading 加 heapq，八十行以内就能得到一个行为可预期的进程内调度器。前提是先想清楚它不负责什么。
先划清能力边界#
要解决的是四个具体问题：
- 触发间隔：30 秒、90 秒这类非整分钟周期，cron 表达不了。
- 重叠执行：上一轮跑了 3 分钟，下一轮又起来，两个线程写同一个临时文件。
- 失败处理：上游偶发 502，等到下个周期太久，立刻重试又会把对方打挂。
- 可观测：日志里要能看出是哪个任务、连续失败了几次。
如果只是「每天凌晨跑一次备份」，cron 配 flock 更省事。进程内调度器的价值在于秒级精度、能复用进程里的连接池和缓存，以及能直接读写内存状态。
顺带说一句为什么不用现成库。APScheduler 功能完整，代价是多一份依赖和一套配置心智；threading.Timer 每次触发都要重建对象，还会随任务数线性堆积线程。如果你的需求就是本文列的四条，自己写反而更容易读懂、更容易改。代码全部用标准库，粘贴到任何 Python 3.9 以上的环境都能跑。
用最小堆管理下一次触发时间#
堆顶永远是最早该执行的任务。每条记录带上 next_run（time.monotonic() 时间戳）和自增的 seq，后者用来打破同一时刻的平局。
这里有个必须知道的细节：dataclass(order=True) 会把所有字段拉进比较，而函数对象不能比大小。给 fn、name 加上 compare=False，否则注册第二个任务时就会抛 TypeError。
# scheduler.py —— 仅标准库，Python 3.9+
import heapq
import logging
import threading
import time
from dataclasses import dataclass, field
log = logging.getLogger("sched")
@dataclass(order=True)
class Job:
next_run: float
seq: int
name: str = field(compare=False)
interval: float = field(compare=False)
fn: object = field(compare=False)
running: bool = field(compare=False, default=False)
failures: int = field(compare=False, default=0)
class Scheduler:
def __init__(self, tick: float = 1.0):
self._heap, self._seq = [], 0
self._tick = tick
self._stop = threading.Event()
self._cv = threading.Condition()
self._thread = threading.Thread(target=self._loop, name="sched", daemon=True)
def every(self, seconds, name=None, catch_up=False):
"""注册固定间隔任务；catch_up=True 表示启动后先立刻跑一次。"""
def register(fn):
with self._cv:
self._seq += 1
first = time.monotonic() + (0.0 if catch_up else float(seconds))
heapq.heappush(self._heap,
Job(first, self._seq, name or fn.__name__, float(seconds), fn))
self._cv.notify()
return fn
return register
注册只做两件事：算出首次触发时间，塞进堆并唤醒正在等待的循环。用 Condition 而不是 time.sleep 轮询，新任务注册后能立刻被感知。
调度循环：三个判断缺一不可#
# 片段：以下方法接在 Scheduler 类内部
def _loop(self):
while True:
with self._cv:
if self._stop.is_set():
return
if not self._heap:
self._cv.wait(self._tick)
continue
job = self._heap[0]
wait = job.next_run - time.monotonic()
if wait > 0:
self._cv.wait(min(wait, self._tick)) # 有新任务注册会被提前唤醒
continue
heapq.heappop(self._heap)
if job.running: # 上一轮没跑完：跳过本次，不排队
job.next_run = time.monotonic() + job.interval
heapq.heappush(self._heap, job)
continue
job.running = True
threading.Thread(target=self._run, args=(job,), daemon=True).start()
def start(self):
self._thread.start()
这三个判断分别对应三类故障：
- 等待要有上限。
wait(min(wait, tick))里的tick是兜底，防止堆顶任务在很远的将来时，注册新任务也无法及时生效；notify()负责让新注册立刻打断等待。 - 运行中直接跳过。不排队是刻意的：如果任务稳定慢于周期，排队只会把内存和时间差一起堆爆，日志里全是过期执行。
- 执行必须离开锁。
_loop只负责派发，真正的函数体放到新线程里跑，否则一个慢任务会把整个调度循环按住。
派发之后的执行体是这样收尾的：
# 片段：以下方法同样接在 Scheduler 类内部
def _run(self, job):
started = time.monotonic()
try:
job.fn()
job.failures = 0
except Exception:
job.failures += 1
log.exception("job %s failed (attempt %d)", job.name, job.failures)
finally:
# 失败退避：间隔 × 2^n，封顶 8 倍；成功后退回原间隔
delay = job.interval * min(2 ** job.failures, 8)
job.running = False
with self._cv:
job.next_run = max(started + delay, time.monotonic())
heapq.heappush(self._heap, job)
self._cv.notify()
def stop(self, timeout: float = 10.0):
self._stop.set()
with self._cv:
self._cv.notify_all()
self._thread.join(timeout)
三个判断之后是清理工作。无论成功还是异常，finally 都要把任务重新压回堆里，否则这个任务从此消失；running=False 也要在这里复位，不然下一轮会被永久跳过。
退避公式是 interval × 2^failures，最长 8 倍。它适合「上游短时间不可用」，不适合「必须准点」的场景——后者应该让任务失败并告警，而不是悄悄推迟。这套策略的代价是延迟不可控：连续失败三次后，一次重试可能排到 40 秒开外，用之前先确认业务能接受。
跑起来#
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s")
sched = Scheduler(tick=0.2)
@sched.every(5, name="sync-ledger", catch_up=True)
def sync_ledger():
log.info("pull ledger ...")
sched.start()
try:
time.sleep(30)
finally:
sched.stop()
启动后会立刻补跑一次，之后每 5 秒触发。catch_up=True 的语义仅仅是「进程起来先跑一次」，不是把停机期间漏掉的周期全部补上——这一点容易被误解。
调试时建议再加一行观测：记录 actual_start - job.next_run 的差值，也就是实际开始时间比计划晚了多少。这个值持续偏大，说明线程池已被占满或者锁竞争严重；偶尔尖刺则属于正常的 GC 和调度抖动。tick 决定了精度上限，默认 1 秒，要做到亚秒级就把它调小，代价是空转次数变多。
stop() 只是不再派发新任务。已经在跑的任务会在 join 超时后被守护线程直接终止，写文件到一半就可能留下半成品。所有任务都应设计成可重入、可重试，落盘用「临时文件 + os.replace」。
上线前还得补三件事#
进程内调度器最大的风险不在代码，在部署形态。
| 场景 | cron + flock | 本文调度器 |
|---|---|---|
| 整点、每天固定时刻 | 合适 | 不擅长，需自己算时间 |
| 秒级、亚分钟间隔 | 做不到 | 合适 |
| 复用进程内连接池 | 做不到 | 合适 |
| 多副本只跑一次 | 需文件锁 | 做不到，必须外部锁 |
| 崩溃后补齐漏跑 | 做不到 | 只能补一次，不能补齐 |
- 多副本：三台机器就是三份执行。用 Redis
SET key val NX EX ttl或数据库行锁，锁的 TTL 要大于单次执行时间，否则锁过期后仍会并发。 - 卡死：Python 线程杀不掉。给任务加「开始时间 + 心跳」，超时就告警；真正的硬超时只能把任务丢进子进程。
-
重启丢失：堆在内存里，重启即清空。要精确补偿，得把
last_run落库，启动时按「漏了几个周期」决定补几次。 -
重活别放进线程：这里的并发模型是「一个任务一个线程」，适合 IO 等待。真要做视频转码、大批量计算，线程会被 GIL 串行化，应该把任务提交给进程池，调度器只负责按时投递。
还有个容易忽略的坑：不要用 time.time() 做调度。NTP 回拨几分钟，调度器就会静默停摆同样长的时间；monotonic() 单调递增，只适合测量间隔，不能直接当墙钟时间写日志。
结论#
- 用
heapq+Condition+ 守护线程，80 行足够覆盖秒级周期的进程内调度。 dataclass(order=True)里的函数字段必须compare=False，这是最先撞上的坑。- 重叠执行选择「跳过」而不是「排队」，失败选择「指数退避」而不是「立即重试」。
- 调度器不解决多副本去重和持久化，这两件事要在部署层补。
- 用
monotonic()计算间隔，用time.time()打日志，两者不要混用。
完