cron 有个默认行为，很多人第一次踩坑才知道：不管上一轮跑没跑完，到点它都会再启动一轮。任务偶尔变慢一次，两轮、三轮就叠在一起抢同一份数据；任务卡住一次，之后每轮都在等一个永远不会结束的进程；而最麻烦的是第三种——任务悄悄失败了，cron 又不像前台程序那样会喊你，你可能几天后才发现数据早断了。
下面这个守门脚本只做三件事：拿不到锁就跳过、超时就整组杀掉、成功就留一张心跳。只用 Python 标准库，Linux 上能跑，改一改就能套在任何已有任务外面，不用改任务本身的代码。
前置知识：会写基础 Python，会用 crontab -e 加一条定时任务，知道绝对路径和相对路径的区别。文中路径都以 /home/user/ 为例，换成你自己的实际路径。
一、锁：上一轮没结束，下一轮直接跳过
防重叠最可靠的办法是文件锁。Python 在 Linux 上可以用 fcntl.flock，以非阻塞方式抢独占锁：抢到了才往下跑，抢不到立刻退出，绝不排队等待。排队是最差的选择——每 5 分钟一轮的任务卡住一夜，能攒出上百个排队进程。
两个细节别搞反：第一，锁文件句柄必须在整个任务期间保持打开，函数一返回、句柄一关，锁就释放了；第二，千万别用删锁文件的方式“清锁”，删掉的只是文件名，持有锁的旧进程还在跑，新进程建个同名文件又能拿到锁，重叠照样发生。真卡住了就杀进程，而不是删文件。
另外要说明适用范围：这是建议锁，只对同样遵守这把锁的进程有效；而且只适用于单台机器，跨机器的任务需要分布式锁，那是另一个话题。
二、超时：杀掉整组进程，不只杀父进程
subprocess.run(timeout=...) 超时后只会处理直接启动的子进程。如果任务又拉起了自己的子进程（比如再调一个脚本或一个网络命令），那些孤儿进程会继续占着资源、写着文件，表面上看任务已经“超时退出了”。正确做法是启动时用 start_new_session=True 让任务自成一个进程组，超时后用 killpg 把整组进程一起杀掉，再收尸。这个坑在真实排障里出现过：超时只杀直接子进程，孤儿进程占住管道，调用方能卡到天荒地老。
超时时间怎么定？取任务平时用时的 3 倍左右，再留一点余量。平时 40 秒跑完的任务，设 240 秒比较合理；设得太短，网络抖动一次就误杀，设得太长，卡住后发现得太晚。
三、心跳：成功才留痕，过期就是报警信号
很多任务“成功时不说话，失败时才写日志”，结果失败和“根本没启动”长得一模一样。反过来做：只有成功跑完才更新心跳文件，里面写成功时间和耗时；跳过和失败都只写一行日志。另起一个很便宜的检查：心跳文件的修改时间如果超过“间隔的两倍加宽限”还没更新，就说明任务要么死了、要么一直在失败，值得去看日志。
日志也别贪多：每次运行只记一行——结果、耗时，失败时附上输出的最后 500 字就够了。完整输出留给任务自己需要时再写，守门脚本的日志是用来一眼定位的，不是用来存档的。
完整代码：套在任何命令外面
把下面代码存为 run_guard.py，用法是把原命令跟在 -- 后面：
#!/usr/bin/env python3
"""cron 守门脚本：防重叠、超时整组杀、成功写心跳。仅 Linux，标准库。"""
import argparse, fcntl, json, os, signal, subprocess, sys, time
from datetime import datetime
def log(msg):
print(f"{datetime.now().astimezone().isoformat(timespec='seconds')} {msg}", flush=True)
def main():
p = argparse.ArgumentParser()
p.add_argument("--lock", required=True, help="锁文件路径")
p.add_argument("--timeout", type=int, default=300, help="超时秒数")
p.add_argument("--heartbeat", required=True, help="成功心跳文件路径")
p.add_argument("cmd", nargs=argparse.REMAINDER)
args = p.parse_args()
cmd = args.cmd[1:] if args.cmd and args.cmd[0] == "--" else args.cmd
if not cmd:
print("用法：run_guard.py --lock 锁文件 --heartbeat 心跳文件 -- 命令...", file=sys.stderr)
return 2
for path in (args.lock, args.heartbeat):
parent = os.path.dirname(path)
if parent:
os.makedirs(parent, exist_ok=True)
lock_f = open(args.lock, "a+")
try:
fcntl.flock(lock_f, fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
log("SKIP busy：上一轮仍在运行，本轮跳过")
return 0
start = time.monotonic()
try:
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
stderr=subprocess.STDOUT, text=True,
start_new_session=True)
try:
out, _ = proc.communicate(timeout=args.timeout)
except subprocess.TimeoutExpired:
os.killpg(proc.pid, signal.SIGKILL)
out, _ = proc.communicate()
log(f"TIMEOUT：超过 {args.timeout} 秒，已整组杀掉")
return 124
spent = time.monotonic() - start
if proc.returncode == 0:
with open(args.heartbeat, "w") as f:
json.dump({"ok_at": datetime.now().astimezone().isoformat(),
"seconds": round(spent, 1)}, f)
log(f"OK：耗时 {spent:.1f} 秒")
return 0
log(f"FAIL：退出码 {proc.returncode}，输出尾部：{out[-500:]}")
return proc.returncode
finally:
lock_f.close()
if __name__ == "__main__":
raise SystemExit(main())接 cron 的写法（每 5 分钟一轮，超时 240 秒）：
*/5 * * * * /usr/bin/python3 /home/user/run_guard.py --lock /home/user/.locks/job.lock --timeout 240 --heartbeat /home/user/.locks/job.ok -- /usr/bin/python3 /home/user/job.py >> /home/user/job.log 2>&1心跳过期检查可以单独放一条 cron，每天跑一次就够了，核心判断只有一句：心跳文件超过 12 分钟（两轮间隔加 2 分钟宽限）没更新，就值得人工看一眼：
find /home/user/.locks/job.ok -mmin +12 2>/dev/null | grep -q . && echo "心跳过期：任务可能已停"上线前先做三个实验
不要直接把生产任务套进去就走，先在测试命令上验证三件事：第一，起一个 30 秒的任务，10 秒内再起一轮，第二轮应该打印 SKIP 并立即退出；第二，起一个 60 秒的任务，把超时设成 2 秒，应该在 2 秒后报 TIMEOUT，且用 ps 确认没有残留子进程；第三，跑一次成功任务，确认心跳文件被更新，再故意让任务返回非零，确认心跳时间不变、日志里有 FAIL 和输出尾部。三项都通过，这个守门脚本才算真的在守门。
关于文件锁的更多细节，可以参考这篇讲非阻塞锁的文章：用文件锁阻止重叠的 cron 任务，以及 如何防止重复的 cron 执行。锁本身不复杂，复杂的是超时和静默失败这两处——而这两处恰恰是只加一把锁解决不了的。