Cron 按日历字段匹配时间，并在匹配时启动命令。它不会天然等待同一个命令的上一次执行结束。如果每分钟触发的任务要运行七十秒，下一分钟就可能再启动一份。因此，触发规则、并发规则和成功完成状态必须分别设计，不能因为写好了一个周期表达式，就认为重复工作的所有行为都已经确定。
把字段理解为日历选择 传统用户 crontab 一行依次包含分钟、小时、月内日期、月份、星期，然后是命令。星号表示该字段的全部取值。分钟位置的 */15 是在每个小时内选择相应分钟值，不是从上次任务结束之后再计时十五分钟。步长如果不能均匀划分字段范围，跨过边界时就可能出现不同长度的间隔，不能把它当作连续经过时长。
Debian/Vixie 格式对于两个日期字段有特殊规则：当月内日期和星期都受到限制时，满足其中一个日期条件就可能运行，而分钟、小时和月份限制仍然必须满足。这不是月份与星期之间的“或”关系。安装简短表达式之前，应列出具有代表性的实际日期核对意图，避免以为“每月某日恰好是周五”却写成两个日期条件各自都能触发。
时区、夏令时变化和停机都会影响日历调度，不同实现对跳过或重复时段的处理并不完全相同。Debian 文档为部分小幅时钟调整和固定时间任务定义了特殊行为，从其他调度器复制结论可能不适用。本实验没有修改时钟，也没有覆盖这些切换情况。日历触发同样不等于持久队列，不能自动保证停机期间错过的每次任务都会补跑。
坏掉的样子 重叠任务可能重复产生副作用、争夺资源或相互覆盖文件。如果要求同一时刻只有一个实例，就要明确实现这一政策。非阻塞建议锁可以让后来的实例记录跳过；阻塞锁则可能留下等待进程。两种选择都不自动提供业务效果恰好一次，也不能修复已经执行了一半的操作。是否允许跳过，应来自任务自己的业务要求。
本地文件锁只协调使用同一相关锁机制的参与者，不会自动串行化无关主机上的工作。网络文件系统的锁语义也需要单独确认。锁必须覆盖实际工作生命周期，跳过结果则应能被观察。如果一个实例永久卡住，后来每次触发都静默丢弃，表面上调度器还在工作，但业务可能早已停止推进，因此还要监控最近成功完成距今多久。
Cron 也并非一律没有日志。Debian 默认通过 syslog 记录任务启动，并提供更细的日志选项。命令输出可能在可用邮件链路存在时被投递，但极简容器经常没有这样的链路。应把启动、完成、失败和跳过结果写到明确目的地。启动记录只能证明任务被派发，不能证明真正的应用工作已经成功完成。
快速验证 使用全新的临时 Debian 12 Docker 容器，以 root 运行，确保没有既有 root 任务表或 cron 进程。通过 apt 安装 python3、cron 和 procps，将下面代码保存为 cron-overlap.py，再执行 python3 cron-overlap.py。需要预留约两分钟，因为程序等待的是实际日历触发，没有用模拟调度器替代 cron，也没有加速或调整宿主时钟。
from pathlib import Path from tempfile import TemporaryDirectory import json import os import subprocess import time assert os.geteuid() == 0 and Path('/.dockerenv').exists(), 'use a disposable root Linux container' assert subprocess.run(['pgrep', '-x', 'cron'], capture_output=True).returncode == 1 assert subprocess.run(['crontab', '-l'], capture_output=True).returncode != 0 with TemporaryDirectory(prefix='cron-overlap-') as directory: root = Path(directory) events_path = root / 'events.jsonl' stop = root / 'stop' worker = root / 'worker.py' worker.write_text('''from pathlib import Path import fcntl,json,os,sys,time root=Path(__file__).parent kind=sys.argv[1] def record(event): data=json.dumps({'event':event,'kind':kind,'pid':os.getpid(),'monotonic':time.monotonic()})+'\\n' fd=os.open(root/'events.jsonl',os.O_WRONLY|os.O_CREAT|os.O_APPEND,0o600) try: os.write(fd,data.encode()) finally: os.close(fd) lock=None if kind=='locked': lock=open(root/'job.lock','a') try: fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB) except BlockingIOError: record('skip') raise SystemExit(0) record('start') try: until=time.monotonic()+70 while time.monotonic()<until and not (root/'stop').exists(): time.sleep(0.1) finally: record('end') if lock: lock.close() ''') table = ''.join(f'* * * * * /usr/bin/python3 {worker} {kind} >>{root}/worker.log 2>&1\n' for kind in ('plain', 'locked')) subprocess.run(['crontab', '-'], input=table, text=True, check=True) daemon = subprocess.Popen(['cron', '-f'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def read_events(): if not events_path.exists(): return [] return [json.loads(line) for line in events_path.read_text().splitlines()] try: deadline = time.monotonic() + 140 while time.monotonic() < deadline: assert daemon.poll() is None, 'cron exited' events = read_events() plain = [e for e in events if (e['kind'], e['event']) == ('plain', 'start')] locked = [e for e in events if (e['kind'], e['event']) == ('locked', 'start')] skipped = [e for e in events if (e['kind'], e['event']) == ('locked', 'skip')] if len(plain) >= 2 and skipped: break time.sleep(0.2) else: raise AssertionError('two real cron ticks were not observed') assert len(plain) == 2 and len(locked) == 1 and len(skipped) == 1 assert plain[1]['monotonic'] - plain[0]['monotonic'] < 70 assert not any(e['event'] == 'end' and e['pid'] == plain[0]['pid'] for e in events) print(json.dumps({'actual_cron_starts': len(plain), 'overlap_observed': True, 'seconds_between_starts': round(plain[1]['monotonic']-plain[0]['monotonic'], 3), 'locked_starts': len(locked), 'locked_skips': len(skipped)}), flush=True) finally: daemon.terminate() stop.touch() daemon.communicate(timeout=5) subprocess.run(['crontab', '-r'], check=True) deadline = time.monotonic() + 5 while time.monotonic() < deadline: events = read_events() started = {e['pid'] for e in events if e['event'] == 'start'} ended = {e['pid'] for e in events if e['event'] == 'end'} if started <= ended: break time.sleep(0.1) else: raise AssertionError('test workers did not finish') 两条每分钟执行的记录会运行原本持续七十秒的工作进程。plain 分支没有锁；locked 分支使用共享文件和非阻塞独占 flock，未取得锁时记录 skip 并退出。每个事件包含进程标识和单调时间，能够直接区分不同实例，并检查第二次启动发生时第一次是否仍在执行，而不是仅根据配置推测并发。
...
