Python 固定间隔任务：用 monotonic 避免累计延迟，跳过错过时段

巡检脚本常写成“执行一次，再 sleep(60)”。如果每次处理花 8 秒，两次启动就相隔约 68 秒，运行越久越偏离原来的节奏。本文面向初学者到中级读者，用 Python 标准库实现固定启动间隔、串行执行、跳过错过时段的小循环。

适合短期监控、轮询本地状态和有限次数的采样。不修改 cron 或系统服务，也不安装依赖。

1. 方法：按计划时刻等待

假设间隔为 1 秒，第一次立即执行。轻量任务应尽量在相对时间 0、1、2 秒启动，而不是每次结束后再等完整的 1 秒。

Python 官方时间文档说明，time.monotonic() 不会倒退，也不受系统时钟调整影响；它的起点没有日历含义，只能用读数之差计算持续时间。记录日期仍用日历时间，不能把 monotonic 读数与文件修改时间相减。

本例的调度规则是：

第一次立即开始；后续计划时刻沿原来的时间轴递增。
一次只执行一个任务；执行完成后，跳过已经过去的计划时刻。
如果恰好赶上计划时刻，允许立即执行；不会补跑更早错过的任务。
--count 是实际调用次数。任务抛出异常就停止，返回非零退出码。

例如任务耗时 2.5 秒、间隔 1 秒时，启动时刻约为 0、3、6 秒。1、2 秒两个时段被跳过。对于“每条记录都必须处理”的队列，应保存待办记录，不能采用这种丢弃过期时段的策略。

2. 准备：建立演示目录

需要 Python 3.9 或更新版本。以下示例已在 Linux、Python 3.11.4 下测试，命令适用于 bash 和 zsh。重复实验时选择新的目录名：

python3 --version
mkdir periodic-demo
cd periodic-demo

把下一节保存为当前目录的 periodic.py。演示任务只打印 JSON，再用等待模拟处理耗时，不访问网络或业务文件。

3. 实现：固定间隔和超时跳过

"""Run a finite, sequential task on a monotonic schedule. Python 3.9+."""
import argparse
import json
import math
import sys
import time


def seconds(value):
    number = float(value)
    if not math.isfinite(number) or not 0 <= number <= 86400:
        raise argparse.ArgumentTypeError("must be finite and between 0 and 86400")
    return number


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def run_periodic(action, interval, count, clock=time.monotonic, sleep=time.sleep):
    deadline = clock()
    for number in range(1, count + 1):
        while True:
            remaining = deadline - clock()
            if remaining <= 0:
                break
            sleep(remaining)
        action(number)
        if number == count:
            break
        deadline += interval
        now = clock()
        if deadline < now:
            missed = math.ceil((now - deadline) / interval)
            deadline += missed * interval


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--interval", type=seconds, default=1.0,
                        help="start interval in seconds, 0.001 to 86400")
    parser.add_argument("--count", type=positive_int, default=3,
                        help="number of actual task executions (default: 3)")
    parser.add_argument("--work-seconds", type=seconds, default=0.25,
                        help="simulated task duration (default: 0.25)")
    parser.add_argument("--fail-at", type=positive_int,
                        help="simulate a failure at this execution number")
    args = parser.parse_args()
    if args.interval < 0.001:
        parser.error("--interval must be at least 0.001 seconds")
    started = time.monotonic()

    def action(number):
        if number == args.fail_at:
            raise RuntimeError("simulated task failure")
        print(json.dumps({"run": number, "elapsed_seconds": round(
            time.monotonic() - started, 3)}), flush=True)
        time.sleep(args.work_seconds)

    try:
        run_periodic(action, args.interval, args.count)
    except KeyboardInterrupt:
        return 130
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

关键在于 deadline += interval：下次时间从原计划递增，不从任务结束时重算。math.ceil() 算出需要跳过多少个时段；while 每次等待后重新计算剩余时间，避免无条件调用负数 sleep()。

命令行把秒数限制为有限值，范围为 0–86400，间隔还要求至少 0.001 秒。这样能拒绝 nan、inf 和负数；0.001 只是输入下限，不代表精度保证。验证方式使用 math.isfinite() 和 argparse 的自定义类型。

接入真实巡检时，替换 action() 中打印和模拟等待的部分，保留失败时抛出异常的行为。如果检查的是报表产物，可以复用文件年龄与空文件检查的判断规则。实际动作若涉及网络或外部命令，需要自行设置超时；这个循环不能中断已经卡住的 action()。

4. 运行：观察快任务和慢任务

查看帮助并运行快任务：

python3 periodic.py --help
python3 periodic.py --interval 1 --work-seconds 0.25 --count 3

典型输出如下，具体耗时会随机器负载变化。每行表示一次任务开始，不是完成：

{"run": 1, "elapsed_seconds": 0.0}
{"run": 2, "elapsed_seconds": 1.0}
{"run": 3, "elapsed_seconds": 2.0}

将模拟耗时改为 2.5 秒：

python3 periodic.py --interval 1 --work-seconds 2.5 --count 3

三次启动应约在 0、3、6 秒，最后一次任务完成后才退出，因此总耗时约 8.5 秒。没有额外的结束后等待，也没有并行补跑。

官方 sleep() 文档指出，系统调度可能使等待时间长于请求值。进程被延迟唤醒时，本例仍会执行一次晚到的任务，再跳过后续已经过期的时段；它不是实时调度器。重启进程会从新的起点开始，不保存以前的进度，也不保证每天某个日历时刻运行。

5. 验证：用模拟时钟检查边界

将下面代码保存为同一目录的 check_periodic.py。它注入一个只推进数值的时钟，立即验证快任务、超时跳过和恰好落在边界的情况，再运行子进程检查退出码：

import subprocess
import sys
from periodic import run_periodic


class FakeClock:
    def __init__(self):
        self.now = 0.0

    def clock(self):
        return self.now

    def sleep(self, seconds):
        assert seconds > 0
        self.now += seconds


for duration, expected in [
    (0.25, [0.0, 1.0, 2.0]),
    (2.5, [0.0, 3.0, 6.0]),
    (2.0, [0.0, 2.0, 4.0]),
]:
    fake = FakeClock()
    starts = []

    def action(number):
        starts.append(fake.now)
        fake.now += duration

    run_periodic(action, 1.0, 3, clock=fake.clock, sleep=fake.sleep)
    assert starts == expected, (duration, starts)

base = [sys.executable, "periodic.py"]
for options, expected_code in [
    (["--interval", "0.01", "--work-seconds", "0", "--count", "2"], 0),
    (["--interval", "0"], 2),
    (["--interval", "nan"], 2),
    (["--interval", "inf"], 2),
    (["--count", "0"], 2),
    (["--work-seconds", "-1"], 2),
    (["--work-seconds", "0", "--fail-at", "1"], 1),
]:
    result = subprocess.run(base + options, capture_output=True, text=True,
                            timeout=5)
    assert result.returncode == expected_code, (options, result)
print("PASS: fast, overrun, exact boundary, CLI errors, task failure")

运行：

python3 check_periodic.py

预期输出：

PASS: fast, overrun, exact boundary, CLI errors, task failure

模拟检查验证调度规则，上一节的真实运行验证接线和输出；两者都不能保证生产机器的唤醒精度。

6. 排错：先区分参数、动作和调度

现象	原因与处理
参数错误，退出码 2	将间隔设为 0.001–86400 秒；计数至少为 1；模拟耗时允许为 0。
`ERROR: simulated task failure`，退出码 1	演示故障生效；真实任务应检查异常原因，修复后再运行。
启动时间偶尔偏晚	查看机器负载和任务耗时；延长间隔，或减少单次工作。
一直没有下一条日志	当前 `action()` 可能未返回；为其内部 I/O 或子进程添加超时。

下面两个命令分别应退出 2 和 1；通过调用方检查退出码，不要把“没有后续输出”当作成功：

python3 periodic.py --interval 0
python3 periodic.py --work-seconds 0 --fail-at 1

正常完成退出 0；终端按 Ctrl+C 退出 130。本文没有安装 SIGTERM 处理器，若要在服务停止时完成当前记录后退出，可参考 Python SIGTERM 收尾处理。若从已有调度入口启动时找不到文件或解释器，按 cron 的 PATH 与工作目录排查固定路径，避免再叠加一个外部重复触发器。

7. 总结

用 monotonic 读数安排相对间隔，并从原计划时刻递增，可以避免把每次处理耗时累计进等待周期。慢任务采用跳过过期时段的明确规则，异常则停止并返回非零状态。先用模拟时钟验证边界，再观察真实启动时间，最后把有超时和错误处理的巡检动作接进去。